Add a MongoDB connector to the exporting engine (#8416)
* Copy files from the MongoDB backend * Update the documentation * Rename functions in the MongoDB backend * Add the connector to the Netdata build * Add an initializer and a worker * Add specific configuration options * Initialize the connector * Add a ring buffer for inserting data to a MongoDB database * Add unit tests
Vladimir Kobal committed
Mar 30, 2020 at 09:54 UTC
36c2e1dbf3d415880d5e5f4ccb4dbf2e882bb8c6
19 files changed
+806
-30
CMakeLists.txt
+31
-4
@@ -662,6 +662,11 @@ set(KINESIS_EXPORTING_FILES
662
exporting/aws_kinesis/aws_kinesis_put_record.h
663
)
664
665
+set(MONGODB_EXPORTING_FILES
666
+ exporting/mongodb/mongodb.c
667
+ exporting/mongodb/mongodb.h
668
+ )
669
+
670
set(KINESIS_BACKEND_FILES
671
backends/aws_kinesis/aws_kinesis.c
672
backends/aws_kinesis/aws_kinesis.h
@@ -782,7 +787,7 @@ ENDIF()
787
IF(libmongoc-1.0_FOUND)
788
message(STATUS "mongodb backend: enabled")
789
785
- list(APPEND NETDATA_FILES ${MONGODB_BACKEND_FILES})
790
+ list(APPEND NETDATA_FILES ${MONGODB_BACKEND_FILES} ${MONGODB_EXPORTING_FILES})
791
list(APPEND NETDATA_COMMON_LIBRARIES ${MONGOC_LIBRARIES})
792
list(APPEND NETDATA_COMMON_INCLUDE_DIRS ${MONGOC_INCLUDE_DIRS})
793
ELSE()
@@ -1033,8 +1038,9 @@ if(BUILD_TESTING)
1038
exporting/tests/system_doubles.c
1039
)
1040
set(TEST_NAME exporting_engine)
1036
- set(KINESIS_LINK_OPTIONS)
1041
set(PROMETHEUS_REMOTE_WRITE_LINK_OPTIONS)
1042
+ set(KINESIS_LINK_OPTIONS)
1043
+ set(MONGODB_LINK_OPTIONS)
1044
if(ENABLE_BACKEND_PROMETHEUS_REMOTE_WRITE)
1045
list(APPEND EXPORTING_ENGINE_FILES ${PROMETHEUS_REMOTE_WRITE_EXPORTING_FILES} ${PROTO_SRCS} ${PROTO_HDRS})
1046
list(
@@ -1054,6 +1060,21 @@ if(ENABLE_BACKEND_KINESIS)
1060
-Wl,--wrap=kinesis_put_record
1061
-Wl,--wrap=kinesis_get_result
1062
)
1063
+endif()
1064
+if(MONGOC_LIBRARIES)
1065
+ list(APPEND EXPORTING_ENGINE_FILES ${MONGODB_EXPORTING_FILES})
1066
+ list(
1067
+ APPEND MONGODB_LINK_OPTIONS
1068
+ -Wl,--wrap=mongoc_init
1069
+ -Wl,--wrap=mongoc_uri_new_with_error
1070
+ -Wl,--wrap=mongoc_uri_get_option_as_int32
1071
+ -Wl,--wrap=mongoc_uri_set_option_as_int32
1072
+ -Wl,--wrap=mongoc_client_new_from_uri
1073
+ -Wl,--wrap=mongoc_client_set_appname
1074
+ -Wl,--wrap=mongoc_client_get_collection
1075
+ -Wl,--wrap=mongoc_uri_destroy
1076
+ -Wl,--wrap=mongoc_collection_insert_many
1077
+ )
1078
endif()
1079
add_executable(${TEST_NAME}_testdriver ${EXPORTING_ENGINE_TEST_FILES} ${EXPORTING_ENGINE_FILES})
1080
target_compile_options(
@@ -1061,7 +1082,12 @@ endif()
1082
PRIVATE
1083
-DUNIT_TESTING
1084
)
1064
- target_include_directories(${TEST_NAME}_testdriver PUBLIC ${CMAKE_CURRENT_BINARY_DIR})
1085
+ target_include_directories(
1086
+ ${TEST_NAME}_testdriver
1087
+ PUBLIC
1088
+ ${CMAKE_CURRENT_BINARY_DIR}
1089
+ ${NETDATA_COMMON_INCLUDE_DIRS}
1090
+ )
1091
target_link_options(
1092
${TEST_NAME}_testdriver
1093
PRIVATE
@@ -1086,8 +1112,9 @@ endif()
1112
-Wl,--wrap=recv
1113
-Wl,--wrap=send
1114
-Wl,--wrap=connect_to_one_of
1089
- ${KINESIS_LINK_OPTIONS}
1115
${PROMETHEUS_REMOTE_WRITE_LINK_OPTIONS}
1116
+ ${KINESIS_LINK_OPTIONS}
1117
+ ${MONGODB_LINK_OPTIONS}
1118
)
1119
target_link_libraries(${TEST_NAME}_testdriver libnetdata ${NETDATA_COMMON_LIBRARIES} ${CMOCKA_LIBRARIES})
1120
add_test(NAME test_${TEST_NAME} COMMAND ${TEST_NAME}_testdriver)
Makefile.am
+27
@@ -519,6 +519,11 @@ KINESIS_EXPORTING_FILES = \
519
exporting/aws_kinesis/aws_kinesis_put_record.h \
520
$(NULL)
521
522
+MONGODB_EXPORTING_FILES = \
523
+ exporting/mongodb/mongodb.c \
524
+ exporting/mongodb/mongodb.h \
525
+ $(NULL)
526
+
527
KINESIS_BACKEND_FILES = \
528
backends/aws_kinesis/aws_kinesis.c \
529
backends/aws_kinesis/aws_kinesis.h \
@@ -765,6 +770,13 @@ exporting/prometheus/remote_write/remote_write.pb.h: exporting/prometheus/remote
770
771
endif
772
773
+if ENABLE_EXPORTING
774
+if ENABLE_BACKEND_MONGODB
775
+ netdata_SOURCES += $(MONGODB_EXPORTING_FILES)
776
+ netdata_LDADD += $(OPTIONAL_MONGOC_LIBS)
777
+endif
778
+endif
779
+
780
if ENABLE_BACKEND_MONGODB
781
netdata_SOURCES += $(MONGODB_BACKEND_FILES)
782
netdata_LDADD += $(OPTIONAL_MONGOC_LIBS)
@@ -902,4 +914,19 @@ if ENABLE_BACKEND_KINESIS
914
-Wl,--wrap=kinesis_get_result \
915
$(NULL)
916
endif
917
+if ENABLE_BACKEND_MONGODB
918
+ exporting_tests_exporting_engine_testdriver_SOURCES += $(MONGODB_EXPORTING_FILES)
919
+ exporting_tests_exporting_engine_testdriver_LDADD += $(OPTIONAL_MONGOC_LIBS)
920
+ exporting_tests_exporting_engine_testdriver_LDFLAGS += \
921
+ -Wl,--wrap=mongoc_init \
922
+ -Wl,--wrap=mongoc_uri_new_with_error \
923
+ -Wl,--wrap=mongoc_uri_get_option_as_int32 \
924
+ -Wl,--wrap=mongoc_uri_set_option_as_int32 \
925
+ -Wl,--wrap=mongoc_client_new_from_uri \
926
+ -Wl,--wrap=mongoc_client_set_appname \
927
+ -Wl,--wrap=mongoc_client_get_collection \
928
+ -Wl,--wrap=mongoc_uri_destroy \
929
+ -Wl,--wrap=mongoc_collection_insert_many \
930
+ $(NULL)
931
+endif
932
endif
backends/backends.c
+4
-4
@@ -596,7 +596,7 @@ void *backends_main(void *ptr) {
596
goto cleanup;
597
}
598
599
- if(likely(!mongodb_init(mongodb_uri, mongodb_database, mongodb_collection, mongodb_default_socket_timeout))) {
599
+ if(likely(!backends_mongodb_init(mongodb_uri, mongodb_database, mongodb_collection, mongodb_default_socket_timeout))) {
600
backend_set_mongodb_variables(&default_port, &backend_response_checker, &backend_request_formatter);
601
do_mongodb = 1;
602
}
@@ -910,10 +910,10 @@ void *backends_main(void *ptr) {
910
while(sent < buffer_len) {
911
const char *first_char = buffer_tostring(b);
912
913
- debug(D_BACKEND, "BACKEND: mongodb_insert(): uri = %s, database = %s, collection = %s, \
913
+ debug(D_BACKEND, "BACKEND: backends_mongodb_insert(): uri = %s, database = %s, collection = %s, \
914
buffer = %zu", mongodb_uri, mongodb_database, mongodb_collection, buffer_len);
915
916
- if(likely(!mongodb_insert((char *)first_char, (size_t)chart_buffered_metrics))) {
916
+ if(likely(!backends_mongodb_insert((char *)first_char, (size_t)chart_buffered_metrics))) {
917
sent += buffer_len;
918
chart_transmission_successes++;
919
chart_receptions++;
@@ -1214,7 +1214,7 @@ cleanup:
1214
1215
#if HAVE_MONGOC
1216
if(do_mongodb) {
1217
- mongodb_cleanup();
1217
+ backends_mongodb_cleanup();
1218
freez(mongodb_uri);
1219
freez(mongodb_database);
1220
freez(mongodb_collection);
backends/mongodb/Makefile.am
-4
@@ -6,7 +6,3 @@ MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
6
dist_noinst_DATA = \
7
README.md \
8
$(NULL)
9
-
10
-dist_libconfig_DATA = \
11
- mongodb.conf \
12
- $(NULL)
backends/mongodb/mongodb.c
+9
-9
@@ -6,10 +6,10 @@
6
7
#define CONFIG_FILE_LINE_MAX ((CONFIG_MAX_NAME + CONFIG_MAX_VALUE + 1024) * 2)
8
9
-mongoc_client_t *mongodb_client;
10
-mongoc_collection_t *mongodb_collection;
9
+static mongoc_client_t *mongodb_client;
10
+static mongoc_collection_t *mongodb_collection;
11
12
-int mongodb_init(const char *uri_string,
12
+int backends_mongodb_init(const char *uri_string,
13
const char *database_string,
14
const char *collection_string,
15
int32_t default_socket_timeout) {
@@ -47,7 +47,7 @@ int mongodb_init(const char *uri_string,
47
return 0;
48
}
49
50
-void free_bson(bson_t **insert, size_t n_documents) {
50
+void backends_free_bson(bson_t **insert, size_t n_documents) {
51
size_t i;
52
53
for(i = 0; i < n_documents; i++)
@@ -56,7 +56,7 @@ void free_bson(bson_t **insert, size_t n_documents) {
56
free(insert);
57
}
58
59
-int mongodb_insert(char *data, size_t n_metrics) {
59
+int backends_mongodb_insert(char *data, size_t n_metrics) {
60
bson_t **insert = calloc(n_metrics, sizeof(bson_t *));
61
bson_error_t error;
62
char *start = data, *end = data;
@@ -77,7 +77,7 @@ int mongodb_insert(char *data, size_t n_metrics) {
77
78
if(unlikely(!insert[n_documents])) {
79
error("BACKEND: %s", error.message);
80
- free_bson(insert, n_documents);
80
+ backends_free_bson(insert, n_documents);
81
return 1;
82
}
83
@@ -88,16 +88,16 @@ int mongodb_insert(char *data, size_t n_metrics) {
88
89
if(unlikely(!mongoc_collection_insert_many(mongodb_collection, (const bson_t **)insert, n_documents, NULL, NULL, &error))) {
90
error("BACKEND: %s", error.message);
91
- free_bson(insert, n_documents);
91
+ backends_free_bson(insert, n_documents);
92
return 1;
93
}
94
95
- free_bson(insert, n_documents);
95
+ backends_free_bson(insert, n_documents);
96
97
return 0;
98
}
99
100
-void mongodb_cleanup() {
100
+void backends_mongodb_cleanup() {
101
mongoc_collection_destroy(mongodb_collection);
102
mongoc_client_destroy(mongodb_client);
103
mongoc_cleanup();
backends/mongodb/mongodb.h
+3
-3
@@ -5,11 +5,11 @@
5
6
#include "backends/backends.h"
7
8
-extern int mongodb_init(const char *uri_string, const char *database_string, const char *collection_string, const int32_t socket_timeout);
8
+extern int backends_mongodb_init(const char *uri_string, const char *database_string, const char *collection_string, const int32_t socket_timeout);
9
10
-extern int mongodb_insert(char *data, size_t n_metrics);
10
+extern int backends_mongodb_insert(char *data, size_t n_metrics);
11
12
-extern void mongodb_cleanup();
12
+extern void backends_mongodb_cleanup();
13
14
extern int read_mongodb_conf(const char *path, char **uri_p, char **database_p, char **collection_p);
15
configure.ac
+1
@@ -1334,6 +1334,7 @@ AC_CONFIG_FILES([
1334
exporting/prometheus/Makefile
1335
exporting/prometheus/remote_write/Makefile
1336
exporting/aws_kinesis/Makefile
1337
+ exporting/mongodb/Makefile
1338
exporting/tests/Makefile
1339
health/Makefile
1340
health/notifications/Makefile
exporting/Makefile.am
+1
@@ -10,6 +10,7 @@ SUBDIRS = \
10
opentsdb \
11
prometheus \
12
aws_kinesis \
13
+ mongodb \
14
$(NULL)
15
16
dist_noinst_DATA = \
exporting/exporting_engine.h
+6
@@ -75,6 +75,11 @@ struct aws_kinesis_specific_config {
75
char *secure_key;
76
};
77
78
+struct mongodb_specific_config {
79
+ char *database;
80
+ char *collection;
81
+};
82
+
83
struct engine_config {
84
const char *prefix;
85
const char *hostname;
@@ -140,6 +145,7 @@ struct engine {
145
time_t now;
146
147
int aws_sdk_initialized;
148
+ int mongoc_initialized;
149
150
struct instance *instance_root;
151
};
exporting/init_connectors.c
+17
-1
@@ -4,8 +4,18 @@
4
#include "graphite/graphite.h"
5
#include "json/json.h"
6
#include "opentsdb/opentsdb.h"
7
-#include "aws_kinesis/aws_kinesis.h"
7
+
8
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
9
#include "prometheus/remote_write/remote_write.h"
10
+#endif
11
+
12
+#if HAVE_KINESIS
13
+#include "aws_kinesis/aws_kinesis.h"
14
+#endif
15
+
16
+#if HAVE_MONGOC
17
+#include "mongodb/mongodb.h"
18
+#endif
19
20
/**
21
* Initialize connectors
@@ -48,6 +58,12 @@ int init_connectors(struct engine *engine)
58
#if HAVE_KINESIS
59
if (init_aws_kinesis_instance(instance) != 0)
60
return 1;
61
+#endif
62
+ break;
63
+ case BACKEND_TYPE_MONGODB:
64
+#if HAVE_MONGOC
65
+ if (init_mongodb_instance(instance) != 0)
66
+ return 1;
67
#endif
68
break;
69
default:
exporting/mongodb/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/mongodb/README.md
new
+31
@@ -0,0 +1,31 @@
1
+# MongoDB exporting connector
2
+
3
+You can use the MongoDB connector and the experimental [exporting engine](../README.md) to archive your agent's metrics
4
+to a MongoDB database for long-term storage, further analysis, or correlation with data from other sources.
5
+
6
+## Prerequisites
7
+
8
+To use MongoDB as an external storage for long-term archiving, you should first
9
+[install](http://mongoc.org/libmongoc/current/installing.html) `libmongoc` 1.7.0 or higher. Next, re-install Netdata
10
+from the source, which detects that the required library is now available.
11
+
12
+## Configuration
13
+
14
+To enable data exporting to a MongoDB database, run `./edit-config exporting.conf`
15
+in the Netdata configuration directory and set the following options:
16
+
17
+```conf
18
+[mongodb:my_instance]
19
+ enabled = yes
20
+ destination = mongodb://<hostname>
21
+ database = your_database_name
22
+ collection = your_collection_name
23
+```
24
+
25
+You can find more information about the `destination` string URI format in the MongoDB
26
+[documentation](https://docs.mongodb.com/manual/reference/connection-string/)
27
+
28
+The default socket timeout depends on the exporting connector update interval. The timeout is 500 ms shorter than the
29
+interval (but not less than 1000 ms). You can alter the timeout using the `sockettimeoutms` MongoDB URI option.
30
+
31
+[](<>)
exporting/mongodb/mongodb.c
new
+313
@@ -0,0 +1,313 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#define EXPORTING_INTERNALS
4
+#include "mongodb.h"
5
+
6
+#define CONFIG_FILE_LINE_MAX ((CONFIG_MAX_NAME + CONFIG_MAX_VALUE + 1024) * 2)
7
+
8
+/**
9
+ * Initialize MongoDB connector specific data, including a ring buffer
10
+ *
11
+ * @param instance an instance data structure.
12
+ * @return Returns 0 on success, 1 on failure.
13
+ */
14
+int mongodb_init(struct instance *instance)
15
+{
16
+ struct mongodb_specific_config *connector_specific_config = instance->config.connector_specific_config;
17
+ mongoc_uri_t *uri;
18
+ bson_error_t bson_error;
19
+
20
+ if (unlikely(!connector_specific_config->collection || !*connector_specific_config->collection)) {
21
+ error("EXPORTING: collection name is a mandatory MongoDB parameter, but it is not configured");
22
+ return 1;
23
+ }
24
+
25
+ uri = mongoc_uri_new_with_error(instance->config.destination, &bson_error);
26
+ if (unlikely(!uri)) {
27
+ error(
28
+ "EXPORTING: failed to parse URI: %s. Error message: %s", instance->config.destination, bson_error.message);
29
+ return 1;
30
+ }
31
+
32
+ int32_t socket_timeout =
33
+ mongoc_uri_get_option_as_int32(uri, MONGOC_URI_SOCKETTIMEOUTMS, instance->config.timeoutms);
34
+ if (!mongoc_uri_set_option_as_int32(uri, MONGOC_URI_SOCKETTIMEOUTMS, socket_timeout)) {
35
+ error("EXPORTING: failed to set %s to the value %d", MONGOC_URI_SOCKETTIMEOUTMS, socket_timeout);
36
+ return 1;
37
+ };
38
+
39
+ struct mongodb_specific_data *connector_specific_data =
40
+ (struct mongodb_specific_data *)instance->connector_specific_data;
41
+
42
+ connector_specific_data->client = mongoc_client_new_from_uri(uri);
43
+ if (unlikely(!connector_specific_data->client)) {
44
+ error("EXPORTING: failed to create a new client");
45
+ return 1;
46
+ }
47
+
48
+ if (!mongoc_client_set_appname(connector_specific_data->client, "netdata")) {
49
+ error("EXPORTING: failed to set client appname");
50
+ };
51
+
52
+ connector_specific_data->collection = mongoc_client_get_collection(
53
+ connector_specific_data->client, connector_specific_config->database, connector_specific_config->collection);
54
+
55
+ mongoc_uri_destroy(uri);
56
+
57
+ // create a ring buffer
58
+ struct bson_buffer *first_buffer = NULL;
59
+
60
+ if (instance->config.buffer_on_failures < 2)
61
+ instance->config.buffer_on_failures = 1;
62
+ else
63
+ instance->config.buffer_on_failures -= 1;
64
+
65
+ for (int i = 0; i < instance->config.buffer_on_failures; i++) {
66
+ struct bson_buffer *current_buffer = callocz(1, sizeof(struct bson_buffer));
67
+
68
+ if (!connector_specific_data->first_buffer)
69
+ first_buffer = current_buffer;
70
+ else
71
+ current_buffer->next = connector_specific_data->first_buffer;
72
+
73
+ connector_specific_data->first_buffer = current_buffer;
74
+ }
75
+
76
+ first_buffer->next = connector_specific_data->first_buffer;
77
+ connector_specific_data->last_buffer = connector_specific_data->first_buffer;
78
+
79
+ return 0;
80
+}
81
+
82
+/**
83
+ * Clean a MongoDB connector instance up
84
+ *
85
+ * @param instance an instance data structure.
86
+ */
87
+void mongodb_cleanup(struct instance *instance)
88
+{
89
+ struct mongodb_specific_data *connector_specific_data =
90
+ (struct mongodb_specific_data *)instance->connector_specific_data;
91
+
92
+ mongoc_collection_destroy(connector_specific_data->collection);
93
+ mongoc_client_destroy(connector_specific_data->client);
94
+
95
+ return;
96
+}
97
+
98
+/**
99
+ * Initialize a MongoDB connector instance
100
+ *
101
+ * @param instance an instance data structure.
102
+ * @return Returns 0 on success, 1 on failure.
103
+ */
104
+int init_mongodb_instance(struct instance *instance)
105
+{
106
+ instance->worker = mongodb_connector_worker;
107
+
108
+ instance->start_batch_formatting = NULL;
109
+ instance->start_host_formatting = format_host_labels_json_plaintext;
110
+ instance->start_chart_formatting = NULL;
111
+
112
+ if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AS_COLLECTED)
113
+ instance->metric_formatting = format_dimension_collected_json_plaintext;
114
+ else
115
+ instance->metric_formatting = format_dimension_stored_json_plaintext;
116
+
117
+ instance->end_chart_formatting = NULL;
118
+ instance->end_host_formatting = flush_host_labels;
119
+ instance->end_batch_formatting = format_batch_mongodb;
120
+
121
+ instance->send_header = NULL;
122
+ instance->check_response = NULL;
123
+
124
+ instance->buffer = (void *)buffer_create(0);
125
+ if (!instance->buffer) {
126
+ error("EXPORTING: cannot create buffer for MongoDB exporting connector instance %s", instance->config.name);
127
+ return 1;
128
+ }
129
+ if (uv_mutex_init(&instance->mutex))
130
+ return 1;
131
+ if (uv_cond_init(&instance->cond_var))
132
+ return 1;
133
+
134
+ struct mongodb_specific_data *connector_specific_data = callocz(1, sizeof(struct mongodb_specific_data));
135
+ instance->connector_specific_data = (void *)connector_specific_data;
136
+
137
+ instance->config.timeoutms =
138
+ (instance->config.update_every >= 2) ? (instance->engine->config.update_every * MSEC_PER_SEC - 500) : 1000;
139
+
140
+ if (!instance->engine->mongoc_initialized) {
141
+ mongoc_init();
142
+ instance->engine->mongoc_initialized = 1;
143
+ }
144
+
145
+ if (unlikely(mongodb_init(instance))) {
146
+ error("EXPORTING: cannot initialize MongoDB exporting connector");
147
+ return 1;
148
+ }
149
+
150
+ return 0;
151
+}
152
+
153
+/**
154
+ * Free an array of BSON structures
155
+ *
156
+ * @param insert an array of documents.
157
+ * @param documents_inserted the number of documents inserted.
158
+ */
159
+void free_bson(bson_t **insert, size_t documents_inserted)
160
+{
161
+ size_t i;
162
+
163
+ for (i = 0; i < documents_inserted; i++)
164
+ bson_destroy(insert[i]);
165
+
166
+ freez(insert);
167
+}
168
+
169
+/**
170
+ * Format a batch for the MongoDB connector
171
+ *
172
+ * @param instance an instance data structure.
173
+ * @return Returns 0 on success, 1 on failure.
174
+ */
175
+int format_batch_mongodb(struct instance *instance)
176
+{
177
+ struct mongodb_specific_data *connector_specific_data =
178
+ (struct mongodb_specific_data *)instance->connector_specific_data;
179
+ struct stats *stats = &instance->stats;
180
+
181
+ bson_t **insert = connector_specific_data->last_buffer->insert;
182
+ if (insert) {
183
+ // ring buffer is full, reuse the oldest element
184
+ connector_specific_data->first_buffer = connector_specific_data->first_buffer->next;
185
+ free_bson(insert, connector_specific_data->last_buffer->documents_inserted);
186
+ }
187
+ insert = callocz((size_t)stats->chart_buffered_metrics, sizeof(bson_t *));
188
+ connector_specific_data->last_buffer->insert = insert;
189
+
190
+ BUFFER *buffer = (BUFFER *)instance->buffer;
191
+ char *start = (char *)buffer_tostring(buffer);
192
+ char *end = start;
193
+
194
+ size_t documents_inserted = 0;
195
+
196
+ while (*end && documents_inserted <= (size_t)stats->chart_buffered_metrics) {
197
+ while (*end && *end != '\n')
198
+ end++;
199
+
200
+ if (likely(*end)) {
201
+ *end = '\0';
202
+ end++;
203
+ } else {
204
+ break;
205
+ }
206
+
207
+ bson_error_t bson_error;
208
+ insert[documents_inserted] = bson_new_from_json((const uint8_t *)start, -1, &bson_error);
209
+
210
+ if (unlikely(!insert[documents_inserted])) {
211
+ error("EXPORTING: %s", bson_error.message);
212
+ free_bson(insert, documents_inserted);
213
+ return 1;
214
+ }
215
+
216
+ start = end;
217
+
218
+ documents_inserted++;
219
+ }
220
+
221
+ buffer_flush(buffer);
222
+
223
+ connector_specific_data->last_buffer->documents_inserted = documents_inserted;
224
+ connector_specific_data->last_buffer = connector_specific_data->last_buffer->next;
225
+
226
+ return 0;
227
+}
228
+
229
+/**
230
+ * MongoDB connector worker
231
+ *
232
+ * Runs in a separate thread for every instance.
233
+ *
234
+ * @param instance_p an instance data structure.
235
+ */
236
+void mongodb_connector_worker(void *instance_p)
237
+{
238
+ struct instance *instance = (struct instance *)instance_p;
239
+ struct mongodb_specific_config *connector_specific_config = instance->config.connector_specific_config;
240
+ struct mongodb_specific_data *connector_specific_data =
241
+ (struct mongodb_specific_data *)instance->connector_specific_data;
242
+
243
+ while (!netdata_exit) {
244
+ struct stats *stats = &instance->stats;
245
+
246
+ uv_mutex_lock(&instance->mutex);
247
+ uv_cond_wait(&instance->cond_var, &instance->mutex);
248
+
249
+ bson_t **insert = connector_specific_data->first_buffer->insert;
250
+ size_t documents_inserted = connector_specific_data->first_buffer->documents_inserted;
251
+
252
+ connector_specific_data->first_buffer->insert = NULL;
253
+ connector_specific_data->first_buffer->documents_inserted = 0;
254
+ connector_specific_data->first_buffer = connector_specific_data->first_buffer->next;
255
+
256
+ uv_mutex_unlock(&instance->mutex);
257
+
258
+ size_t data_size = 0;
259
+ for (size_t i = 0; i < documents_inserted; i++) {
260
+ data_size += insert[i]->len;
261
+ }
262
+
263
+ debug(
264
+ D_BACKEND,
265
+ "EXPORTING: mongodb_insert(): destination = %s, database = %s, collection = %s, data size = %zu",
266
+ instance->config.destination,
267
+ connector_specific_config->database,
268
+ connector_specific_config->collection,
269
+ data_size);
270
+
271
+ if (unlikely(documents_inserted == 0))
272
+ continue;
273
+
274
+ bson_error_t bson_error;
275
+ if (likely(mongoc_collection_insert_many(
276
+ connector_specific_data->collection,
277
+ (const bson_t **)insert,
278
+ documents_inserted,
279
+ NULL,
280
+ NULL,
281
+ &bson_error))) {
282
+ stats->chart_sent_bytes += data_size;
283
+ stats->chart_transmission_successes++;
284
+ stats->chart_receptions++;
285
+ } else {
286
+ // oops! we couldn't send (all or some of the) data
287
+ error("EXPORTING: %s", bson_error.message);
288
+ error(
289
+ "EXPORTING: failed to write data to the database '%s'. "
290
+ "Willing to write %zu bytes, wrote %zu bytes.",
291
+ instance->config.destination, data_size, 0UL);
292
+
293
+ stats->chart_transmission_failures++;
294
+ stats->chart_data_lost_events++;
295
+ stats->chart_lost_bytes += data_size;
296
+ stats->chart_lost_metrics += stats->chart_buffered_metrics;
297
+ }
298
+
299
+ free_bson(insert, documents_inserted);
300
+
301
+ if (unlikely(netdata_exit))
302
+ break;
303
+
304
+ stats->chart_sent_bytes += data_size;
305
+ stats->chart_sent_metrics = stats->chart_buffered_metrics;
306
+
307
+#ifdef UNIT_TESTING
308
+ break;
309
+#endif
310
+ }
311
+
312
+ mongodb_cleanup(instance);
313
+}
exporting/mongodb/mongodb.h
new
+33
@@ -0,0 +1,33 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_EXPORTING_MONGODB_H
4
+#define NETDATA_EXPORTING_MONGODB_H
5
+
6
+#include "exporting/exporting_engine.h"
7
+#include "exporting/json/json.h"
8
+#include <mongoc.h>
9
+
10
+struct bson_buffer {
11
+ bson_t **insert;
12
+ size_t documents_inserted;
13
+
14
+ struct bson_buffer *next;
15
+};
16
+
17
+struct mongodb_specific_data {
18
+ mongoc_client_t *client;
19
+ mongoc_collection_t *collection;
20
+
21
+ bson_t **current_insert;
22
+ struct bson_buffer *first_buffer;
23
+ struct bson_buffer *last_buffer;
24
+};
25
+
26
+int mongodb_init(struct instance *instance);
27
+void mongodb_cleanup(struct instance *instance);
28
+
29
+int init_mongodb_instance(struct instance *instance);
30
+int format_batch_mongodb(struct instance *instance);
31
+void mongodb_connector_worker(void *instance_p);
32
+
33
+#endif //NETDATA_EXPORTING_MONGODB_H
exporting/process_data.c
+4
-4
@@ -329,11 +329,11 @@ int end_batch_formatting(struct engine *engine)
329
*/
330
int prepare_buffers(struct engine *engine)
331
{
332
- netdata_thread_disable_cancelability();
333
- rrd_rdlock();
332
if (start_batch_formatting(engine) != 0)
333
return 1;
334
335
+ netdata_thread_disable_cancelability();
336
+ rrd_rdlock();
337
RRDHOST *host;
338
rrdhost_foreach_read(host)
339
{
@@ -363,11 +363,11 @@ int prepare_buffers(struct engine *engine)
363
return 1;
364
rrdhost_unlock(host);
365
}
366
+ rrd_unlock();
367
+ netdata_thread_enable_cancelability();
368
369
if (end_batch_formatting(engine) != 0)
370
return 1;
369
- rrd_unlock();
370
- netdata_thread_enable_cancelability();
371
372
return 0;
373
}
exporting/read_config.c
+20
@@ -281,6 +281,13 @@ struct engine *read_exporting_config()
281
}
282
#endif
283
284
+#ifndef HAVE_MONGOC
285
+ if (tmp_ci_list->backend_type == BACKEND_TYPE_MONGODB) {
286
+ error("MongoDB support isn't compiled");
287
+ goto next_connector_instance;
288
+ }
289
+#endif
290
+
291
tmp_instance = (struct instance *)callocz(1, sizeof(struct instance));
292
tmp_instance->next = engine->instance_root;
293
engine->instance_root = tmp_instance;
@@ -362,6 +369,19 @@ struct engine *read_exporting_config()
369
instance_name, "aws_secret_access_key", ""));
370
}
371
372
+ if (tmp_instance->config.type == BACKEND_TYPE_MONGODB) {
373
+ struct mongodb_specific_config *connector_specific_config =
374
+ callocz(1, sizeof(struct mongodb_specific_config));
375
+
376
+ tmp_instance->config.connector_specific_config = connector_specific_config;
377
+
378
+ connector_specific_config->database = strdupz(exporter_get(
379
+ instance_name, "database", ""));
380
+
381
+ connector_specific_config->collection = strdupz(exporter_get(
382
+ instance_name, "collection", ""));
383
+ }
384
+
385
#ifdef NETDATA_INTERNAL_CHECKS
386
info(
387
" Dest=[%s], upd=[%d], buffer=[%d] timeout=[%ld] options=[%u]",
exporting/tests/exporting_doubles.c
+83
-1
@@ -244,4 +244,86 @@ int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, siz
244
check_expected_ptr(lost_bytes);
245
return mock_type(int);
246
}
247
-#endif //HAVE_KINESIS
247
+#endif // HAVE_KINESIS
248
+
249
+#if HAVE_MONGOC
250
+void __wrap_mongoc_init()
251
+{
252
+ function_called();
253
+}
254
+
255
+mongoc_uri_t * __wrap_mongoc_uri_new_with_error (const char *uri_string, bson_error_t *error)
256
+{
257
+ function_called();
258
+ check_expected_ptr(uri_string);
259
+ check_expected_ptr(error);
260
+ return mock_ptr_type(mongoc_uri_t *);
261
+}
262
+
263
+int32_t __wrap_mongoc_uri_get_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t fallback)
264
+{
265
+ function_called();
266
+ check_expected_ptr(uri);
267
+ check_expected_ptr(option);
268
+ check_expected(fallback);
269
+ return mock_type(int32_t);
270
+}
271
+
272
+bool __wrap_mongoc_uri_set_option_as_int32 (const mongoc_uri_t *uri, const char *option, int32_t value)
273
+{
274
+ function_called();
275
+ check_expected_ptr(uri);
276
+ check_expected_ptr(option);
277
+ check_expected(value);
278
+ return mock_type(bool);
279
+}
280
+
281
+mongoc_client_t * __wrap_mongoc_client_new_from_uri (const mongoc_uri_t *uri)
282
+{
283
+ function_called();
284
+ check_expected_ptr(uri);
285
+ return mock_ptr_type(mongoc_client_t *);
286
+}
287
+
288
+bool __wrap_mongoc_client_set_appname (mongoc_client_t *client, const char *appname)
289
+{
290
+ function_called();
291
+ check_expected_ptr(client);
292
+ check_expected_ptr(appname);
293
+ return mock_type(bool);
294
+}
295
+
296
+mongoc_collection_t *
297
+__wrap_mongoc_client_get_collection(mongoc_client_t *client, const char *db, const char *collection)
298
+{
299
+ function_called();
300
+ check_expected_ptr(client);
301
+ check_expected_ptr(db);
302
+ check_expected_ptr(collection);
303
+ return mock_ptr_type(mongoc_collection_t *);
304
+}
305
+
306
+void __wrap_mongoc_uri_destroy (mongoc_uri_t *uri)
307
+{
308
+ function_called();
309
+ check_expected_ptr(uri);
310
+}
311
+
312
+bool __wrap_mongoc_collection_insert_many(
313
+ mongoc_collection_t *collection,
314
+ const bson_t **documents,
315
+ size_t n_documents,
316
+ const bson_t *opts,
317
+ bson_t *reply,
318
+ bson_error_t *error)
319
+{
320
+ function_called();
321
+ check_expected_ptr(collection);
322
+ check_expected_ptr(documents);
323
+ check_expected(n_documents);
324
+ check_expected_ptr(opts);
325
+ check_expected_ptr(reply);
326
+ check_expected_ptr(error);
327
+ return mock_type(bool);
328
+}
329
+#endif // HAVE_MONGOC
exporting/tests/test_exporting_engine.c
+186
@@ -977,6 +977,7 @@ static void test_init_aws_kinesis_instance(void **state)
977
expect_string(__wrap_kinesis_init, access_key_id, "test_auth_key_id");
978
expect_string(__wrap_kinesis_init, secret_key, "test_secure_key");
979
expect_value(__wrap_kinesis_init, timeout, 10000);
980
+
981
assert_int_equal(init_aws_kinesis_instance(instance), 0);
982
983
assert_ptr_equal(instance->worker, aws_kinesis_connector_worker);
@@ -1066,6 +1067,178 @@ static void test_aws_kinesis_connector_worker(void **state)
1067
}
1068
#endif // HAVE_KINESIS
1069
1070
+#if HAVE_MONGOC
1071
+static void test_init_mongodb_instance(void **state)
1072
+{
1073
+ struct engine *engine = *state;
1074
+ struct instance *instance = engine->instance_root;
1075
+
1076
+ instance->config.options = EXPORTING_SOURCE_DATA_AS_COLLECTED | EXPORTING_OPTION_SEND_NAMES;
1077
+
1078
+ struct mongodb_specific_config *connector_specific_config = callocz(1, sizeof(struct mongodb_specific_config));
1079
+ instance->config.connector_specific_config = connector_specific_config;
1080
+ connector_specific_config->database = strdupz("test_database");
1081
+ connector_specific_config->collection = strdupz("test_collection");
1082
+ instance->config.buffer_on_failures = 10;
1083
+
1084
+ expect_function_call(__wrap_mongoc_init);
1085
+ expect_function_call(__wrap_mongoc_uri_new_with_error);
1086
+ expect_string(__wrap_mongoc_uri_new_with_error, uri_string, "localhost");
1087
+ expect_not_value(__wrap_mongoc_uri_new_with_error, error, NULL);
1088
+ will_return(__wrap_mongoc_uri_new_with_error, 0xf1);
1089
+
1090
+ expect_function_call(__wrap_mongoc_uri_get_option_as_int32);
1091
+ expect_value(__wrap_mongoc_uri_get_option_as_int32, uri, 0xf1);
1092
+ expect_string(__wrap_mongoc_uri_get_option_as_int32, option, MONGOC_URI_SOCKETTIMEOUTMS);
1093
+ expect_value(__wrap_mongoc_uri_get_option_as_int32, fallback, 1000);
1094
+ will_return(__wrap_mongoc_uri_get_option_as_int32, 1000);
1095
+
1096
+ expect_function_call(__wrap_mongoc_uri_set_option_as_int32);
1097
+ expect_value(__wrap_mongoc_uri_set_option_as_int32, uri, 0xf1);
1098
+ expect_string(__wrap_mongoc_uri_set_option_as_int32, option, MONGOC_URI_SOCKETTIMEOUTMS);
1099
+ expect_value(__wrap_mongoc_uri_set_option_as_int32, value, 1000);
1100
+ will_return(__wrap_mongoc_uri_set_option_as_int32, true);
1101
+
1102
+ expect_function_call(__wrap_mongoc_client_new_from_uri);
1103
+ expect_value(__wrap_mongoc_client_new_from_uri, uri, 0xf1);
1104
+ will_return(__wrap_mongoc_client_new_from_uri, 0xf2);
1105
+
1106
+ expect_function_call(__wrap_mongoc_client_set_appname);
1107
+ expect_value(__wrap_mongoc_client_set_appname, client, 0xf2);
1108
+ expect_string(__wrap_mongoc_client_set_appname, appname, "netdata");
1109
+ will_return(__wrap_mongoc_client_set_appname, true);
1110
+
1111
+ expect_function_call(__wrap_mongoc_client_get_collection);
1112
+ expect_value(__wrap_mongoc_client_get_collection, client, 0xf2);
1113
+ expect_string(__wrap_mongoc_client_get_collection, db, "test_database");
1114
+ expect_string(__wrap_mongoc_client_get_collection, collection, "test_collection");
1115
+ will_return(__wrap_mongoc_client_get_collection, 0xf3);
1116
+
1117
+ expect_function_call(__wrap_mongoc_uri_destroy);
1118
+ expect_value(__wrap_mongoc_uri_destroy, uri, 0xf1);
1119
+
1120
+ assert_int_equal(init_mongodb_instance(instance), 0);
1121
+
1122
+ assert_ptr_equal(instance->worker, mongodb_connector_worker);
1123
+ assert_ptr_equal(instance->start_batch_formatting, NULL);
1124
+ assert_ptr_equal(instance->start_host_formatting, format_host_labels_json_plaintext);
1125
+ assert_ptr_equal(instance->start_chart_formatting, NULL);
1126
+ assert_ptr_equal(instance->metric_formatting, format_dimension_collected_json_plaintext);
1127
+ assert_ptr_equal(instance->end_chart_formatting, NULL);
1128
+ assert_ptr_equal(instance->end_host_formatting, flush_host_labels);
1129
+ assert_ptr_equal(instance->end_batch_formatting, format_batch_mongodb);
1130
+ assert_ptr_equal(instance->send_header, NULL);
1131
+ assert_ptr_equal(instance->check_response, NULL);
1132
+
1133
+ assert_ptr_not_equal(instance->buffer, NULL);
1134
+ buffer_free(instance->buffer);
1135
+
1136
+ assert_ptr_not_equal(instance->connector_specific_data, NULL);
1137
+
1138
+ struct mongodb_specific_data *connector_specific_data =
1139
+ (struct mongodb_specific_data *)instance->connector_specific_data;
1140
+ size_t number_of_buffers = 1;
1141
+ struct bson_buffer *current_buffer = connector_specific_data->first_buffer;
1142
+ while (current_buffer->next != connector_specific_data->first_buffer) {
1143
+ current_buffer = current_buffer->next;
1144
+ number_of_buffers++;
1145
+ if (number_of_buffers == (size_t)(instance->config.buffer_on_failures + 1)) {
1146
+ number_of_buffers = 0;
1147
+ break;
1148
+ }
1149
+ }
1150
+ assert_int_equal(number_of_buffers, 9);
1151
+
1152
+ free(connector_specific_config->database);
1153
+ free(connector_specific_config->collection);
1154
+}
1155
+
1156
+static void test_format_batch_mongodb(void **state)
1157
+{
1158
+ struct engine *engine = *state;
1159
+ struct instance *instance = engine->instance_root;
1160
+ struct stats *stats = &instance->stats;
1161
+
1162
+ struct mongodb_specific_data *connector_specific_data = mallocz(sizeof(struct mongodb_specific_data));
1163
+ instance->connector_specific_data = (void *)connector_specific_data;
1164
+
1165
+ struct bson_buffer *current_buffer = callocz(1, sizeof(struct bson_buffer));
1166
+ connector_specific_data->first_buffer = current_buffer;
1167
+ connector_specific_data->first_buffer->next = current_buffer;
1168
+ connector_specific_data->last_buffer = current_buffer;
1169
+
1170
+ BUFFER *buffer = buffer_create(0);
1171
+ buffer_sprintf(buffer, "{ \"metric\": \"test_metric\" }\n");
1172
+ instance->buffer = buffer;
1173
+ stats->chart_buffered_metrics = 1;
1174
+
1175
+ assert_int_equal(format_batch_mongodb(instance), 0);
1176
+
1177
+ assert_int_equal(connector_specific_data->last_buffer->documents_inserted, 1);
1178
+ assert_int_equal(buffer_strlen(buffer), 0);
1179
+
1180
+ size_t len;
1181
+ char *str = bson_as_canonical_extended_json(connector_specific_data->last_buffer->insert[0], &len);
1182
+ assert_string_equal(str, "{ \"metric\" : \"test_metric\" }");
1183
+
1184
+ freez(str);
1185
+ buffer_free(buffer);
1186
+}
1187
+
1188
+static void test_mongodb_connector_worker(void **state)
1189
+{
1190
+ struct engine *engine = *state;
1191
+ struct instance *instance = engine->instance_root;
1192
+
1193
+ struct mongodb_specific_config *connector_specific_config = callocz(1, sizeof(struct mongodb_specific_config));
1194
+ instance->config.connector_specific_config = connector_specific_config;
1195
+ connector_specific_config->database = strdupz("test_database");
1196
+
1197
+ struct mongodb_specific_data *connector_specific_data = callocz(1, sizeof(struct mongodb_specific_data));
1198
+ instance->connector_specific_data = (void *)connector_specific_data;
1199
+ connector_specific_config->collection = strdupz("test_collection");
1200
+
1201
+ struct bson_buffer *buffer = callocz(1, sizeof(struct bson_buffer));
1202
+ buffer->documents_inserted = 1;
1203
+ connector_specific_data->first_buffer = buffer;
1204
+ connector_specific_data->first_buffer->next = buffer;
1205
+
1206
+ connector_specific_data->first_buffer->insert = callocz(1, sizeof(bson_t *));
1207
+ bson_error_t bson_error;
1208
+ connector_specific_data->first_buffer->insert[0] =
1209
+ bson_new_from_json((const uint8_t *)"{ \"test_key\" : \"test_value\" }", -1, &bson_error);
1210
+
1211
+ connector_specific_data->client = mongoc_client_new("mongodb://localhost");
1212
+ connector_specific_data->collection =
1213
+ __real_mongoc_client_get_collection(connector_specific_data->client, "test_database", "test_collection");
1214
+
1215
+ expect_function_call(__wrap_mongoc_collection_insert_many);
1216
+ expect_value(__wrap_mongoc_collection_insert_many, collection, connector_specific_data->collection);
1217
+ expect_value(__wrap_mongoc_collection_insert_many, documents, connector_specific_data->first_buffer->insert);
1218
+ expect_value(__wrap_mongoc_collection_insert_many, n_documents, 1);
1219
+ expect_value(__wrap_mongoc_collection_insert_many, opts, NULL);
1220
+ expect_value(__wrap_mongoc_collection_insert_many, reply, NULL);
1221
+ expect_not_value(__wrap_mongoc_collection_insert_many, error, NULL);
1222
+ will_return(__wrap_mongoc_collection_insert_many, true);
1223
+
1224
+ mongodb_connector_worker(instance);
1225
+
1226
+ assert_ptr_equal(connector_specific_data->first_buffer->insert, NULL);
1227
+ assert_int_equal(connector_specific_data->first_buffer->documents_inserted, 0);
1228
+ assert_ptr_equal(connector_specific_data->first_buffer, connector_specific_data->first_buffer->next);
1229
+
1230
+ struct stats *stats = &instance->stats;
1231
+ assert_int_equal(stats->chart_sent_bytes, 60);
1232
+ assert_int_equal(stats->chart_transmission_successes, 1);
1233
+ assert_int_equal(stats->chart_receptions, 1);
1234
+ assert_int_equal(stats->chart_sent_bytes, 60);
1235
+ assert_int_equal(stats->chart_sent_metrics, 0);
1236
+
1237
+ free(connector_specific_config->database);
1238
+ free(connector_specific_config->collection);
1239
+}
1240
+#endif // HAVE_MONGOC
1241
+
1242
int main(void)
1243
{
1244
const struct CMUnitTest tests[] = {
@@ -1168,5 +1341,18 @@ int main(void)
1341
test_res += cmocka_run_group_tests_name("kinesis_exporting_connector", kinesis_tests, NULL, NULL);
1342
#endif
1343
1344
+#if HAVE_MONGOC
1345
+ const struct CMUnitTest mongodb_tests[] = {
1346
+ cmocka_unit_test_setup_teardown(
1347
+ test_init_mongodb_instance, setup_configured_engine, teardown_configured_engine),
1348
+ cmocka_unit_test_setup_teardown(
1349
+ test_format_batch_mongodb, setup_configured_engine, teardown_configured_engine),
1350
+ cmocka_unit_test_setup_teardown(
1351
+ test_mongodb_connector_worker, setup_configured_engine, teardown_configured_engine),
1352
+ };
1353
+
1354
+ test_res += cmocka_run_group_tests_name("mongodb_exporting_connector", mongodb_tests, NULL, NULL);
1355
+#endif
1356
+
1357
return test_res;
1358
}
exporting/tests/test_exporting_engine.h
+29
@@ -9,8 +9,18 @@
9
#include "exporting/graphite/graphite.h"
10
#include "exporting/json/json.h"
11
#include "exporting/opentsdb/opentsdb.h"
12
+
13
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
14
#include "exporting/prometheus/remote_write/remote_write.h"
15
+#endif
16
+
17
+#if HAVE_KINESIS
18
#include "exporting/aws_kinesis/aws_kinesis.h"
19
+#endif
20
+
21
+#if HAVE_MONGOC
22
+#include "exporting/mongodb/mongodb.h"
23
+#endif
24
25
#include <stdarg.h>
26
#include <stddef.h>
@@ -128,6 +138,25 @@ void __wrap_kinesis_put_record(
138
size_t data_len);
139
int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, size_t *sent_bytes, size_t *lost_bytes);
140
141
+void __wrap_mongoc_init();
142
+mongoc_uri_t *__wrap_mongoc_uri_new_with_error(const char *uri_string, bson_error_t *error);
143
+int32_t __wrap_mongoc_uri_get_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t fallback);
144
+bool __wrap_mongoc_uri_set_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t value);
145
+mongoc_client_t *__wrap_mongoc_client_new_from_uri(const mongoc_uri_t *uri);
146
+bool __wrap_mongoc_client_set_appname(mongoc_client_t *client, const char *appname);
147
+mongoc_collection_t *
148
+__wrap_mongoc_client_get_collection(mongoc_client_t *client, const char *db, const char *collection);
149
+mongoc_collection_t *
150
+__real_mongoc_client_get_collection(mongoc_client_t *client, const char *db, const char *collection);
151
+void __wrap_mongoc_uri_destroy(mongoc_uri_t *uri);
152
+bool __wrap_mongoc_collection_insert_many(
153
+ mongoc_collection_t *collection,
154
+ const bson_t **documents,
155
+ size_t n_documents,
156
+ const bson_t *opts,
157
+ bson_t *reply,
158
+ bson_error_t *error);
159
+
160
// -----------------------------------------------------------------------
161
// fixtures
162