Add a MongoDB connector to the exporting engine (netdata#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

CMakeLists.txt

@@ -662,6 +662,11 @@ set(KINESIS_EXPORTING_FILES
         exporting/aws_kinesis/aws_kinesis_put_record.h
         )
 
+set(MONGODB_EXPORTING_FILES
+        exporting/mongodb/mongodb.c
+        exporting/mongodb/mongodb.h
+        )
+
 set(KINESIS_BACKEND_FILES
         backends/aws_kinesis/aws_kinesis.c
         backends/aws_kinesis/aws_kinesis.h
@@ -782,7 +787,7 @@ ENDIF()
 IF(libmongoc-1.0_FOUND)
     message(STATUS "mongodb backend: enabled")
 
-    list(APPEND NETDATA_FILES ${MONGODB_BACKEND_FILES})
+    list(APPEND NETDATA_FILES ${MONGODB_BACKEND_FILES} ${MONGODB_EXPORTING_FILES})
     list(APPEND NETDATA_COMMON_LIBRARIES ${MONGOC_LIBRARIES})
     list(APPEND NETDATA_COMMON_INCLUDE_DIRS ${MONGOC_INCLUDE_DIRS})
 ELSE()
@@ -1033,8 +1038,9 @@ if(BUILD_TESTING)
         exporting/tests/system_doubles.c
         )
     set(TEST_NAME exporting_engine)
-    set(KINESIS_LINK_OPTIONS)
     set(PROMETHEUS_REMOTE_WRITE_LINK_OPTIONS)
+    set(KINESIS_LINK_OPTIONS)
+    set(MONGODB_LINK_OPTIONS)
 if(ENABLE_BACKEND_PROMETHEUS_REMOTE_WRITE)
     list(APPEND EXPORTING_ENGINE_FILES ${PROMETHEUS_REMOTE_WRITE_EXPORTING_FILES} ${PROTO_SRCS} ${PROTO_HDRS})
     list(
@@ -1054,14 +1060,34 @@ if(ENABLE_BACKEND_KINESIS)
         -Wl,--wrap=kinesis_put_record
         -Wl,--wrap=kinesis_get_result
     )
+endif()
+if(MONGOC_LIBRARIES)
+    list(APPEND EXPORTING_ENGINE_FILES ${MONGODB_EXPORTING_FILES})
+    list(
+        APPEND MONGODB_LINK_OPTIONS
+        -Wl,--wrap=mongoc_init
+        -Wl,--wrap=mongoc_uri_new_with_error
+        -Wl,--wrap=mongoc_uri_get_option_as_int32
+        -Wl,--wrap=mongoc_uri_set_option_as_int32
+        -Wl,--wrap=mongoc_client_new_from_uri
+        -Wl,--wrap=mongoc_client_set_appname
+        -Wl,--wrap=mongoc_client_get_collection
+        -Wl,--wrap=mongoc_uri_destroy
+        -Wl,--wrap=mongoc_collection_insert_many
+    )
 endif()
     add_executable(${TEST_NAME}_testdriver ${EXPORTING_ENGINE_TEST_FILES} ${EXPORTING_ENGINE_FILES})
     target_compile_options(
         ${TEST_NAME}_testdriver
         PRIVATE
         -DUNIT_TESTING
     )
-    target_include_directories(${TEST_NAME}_testdriver PUBLIC ${CMAKE_CURRENT_BINARY_DIR})
+    target_include_directories(
+        ${TEST_NAME}_testdriver
+        PUBLIC
+        ${CMAKE_CURRENT_BINARY_DIR}
+        ${NETDATA_COMMON_INCLUDE_DIRS}
+    )
     target_link_options(
         ${TEST_NAME}_testdriver
         PRIVATE
@@ -1086,8 +1112,9 @@ endif()
         -Wl,--wrap=recv
         -Wl,--wrap=send
         -Wl,--wrap=connect_to_one_of
-        ${KINESIS_LINK_OPTIONS}
         ${PROMETHEUS_REMOTE_WRITE_LINK_OPTIONS}
+        ${KINESIS_LINK_OPTIONS}
+        ${MONGODB_LINK_OPTIONS}
     )
     target_link_libraries(${TEST_NAME}_testdriver libnetdata ${NETDATA_COMMON_LIBRARIES} ${CMOCKA_LIBRARIES})
     add_test(NAME test_${TEST_NAME} COMMAND ${TEST_NAME}_testdriver)

Makefile.am

@@ -519,6 +519,11 @@ KINESIS_EXPORTING_FILES = \
     exporting/aws_kinesis/aws_kinesis_put_record.h \
     $(NULL)
 
+MONGODB_EXPORTING_FILES = \
+    exporting/mongodb/mongodb.c \
+    exporting/mongodb/mongodb.h \
+    $(NULL)
+
 KINESIS_BACKEND_FILES = \
     backends/aws_kinesis/aws_kinesis.c \
     backends/aws_kinesis/aws_kinesis.h \
@@ -765,6 +770,13 @@ exporting/prometheus/remote_write/remote_write.pb.h: exporting/prometheus/remote
 
 endif
 
+if ENABLE_EXPORTING
+if ENABLE_BACKEND_MONGODB
+    netdata_SOURCES += $(MONGODB_EXPORTING_FILES)
+    netdata_LDADD += $(OPTIONAL_MONGOC_LIBS)
+endif
+endif
+
 if ENABLE_BACKEND_MONGODB
     netdata_SOURCES += $(MONGODB_BACKEND_FILES)
     netdata_LDADD += $(OPTIONAL_MONGOC_LIBS)
@@ -902,4 +914,19 @@ if ENABLE_BACKEND_KINESIS
         -Wl,--wrap=kinesis_get_result \
         $(NULL)
 endif
+if ENABLE_BACKEND_MONGODB
+    exporting_tests_exporting_engine_testdriver_SOURCES += $(MONGODB_EXPORTING_FILES)
+    exporting_tests_exporting_engine_testdriver_LDADD += $(OPTIONAL_MONGOC_LIBS)
+    exporting_tests_exporting_engine_testdriver_LDFLAGS += \
+        -Wl,--wrap=mongoc_init \
+        -Wl,--wrap=mongoc_uri_new_with_error \
+        -Wl,--wrap=mongoc_uri_get_option_as_int32 \
+        -Wl,--wrap=mongoc_uri_set_option_as_int32 \
+        -Wl,--wrap=mongoc_client_new_from_uri \
+        -Wl,--wrap=mongoc_client_set_appname \
+        -Wl,--wrap=mongoc_client_get_collection \
+        -Wl,--wrap=mongoc_uri_destroy \
+        -Wl,--wrap=mongoc_collection_insert_many \
+        $(NULL)
+endif
 endif

backends/backends.c

@@ -596,7 +596,7 @@ void *backends_main(void *ptr) {
                 goto cleanup;
             }
 
-            if(likely(!mongodb_init(mongodb_uri, mongodb_database, mongodb_collection, mongodb_default_socket_timeout))) {
+            if(likely(!backends_mongodb_init(mongodb_uri, mongodb_database, mongodb_collection, mongodb_default_socket_timeout))) {
                 backend_set_mongodb_variables(&default_port, &backend_response_checker, &backend_request_formatter);
                 do_mongodb = 1;
             }
@@ -910,10 +910,10 @@ void *backends_main(void *ptr) {
             while(sent < buffer_len) {
                 const char *first_char = buffer_tostring(b);
 
-                debug(D_BACKEND, "BACKEND: mongodb_insert(): uri = %s, database = %s, collection = %s, \
+                debug(D_BACKEND, "BACKEND: backends_mongodb_insert(): uri = %s, database = %s, collection = %s, \
                       buffer = %zu", mongodb_uri, mongodb_database, mongodb_collection, buffer_len);
 
-                if(likely(!mongodb_insert((char *)first_char, (size_t)chart_buffered_metrics))) {
+                if(likely(!backends_mongodb_insert((char *)first_char, (size_t)chart_buffered_metrics))) {
                     sent += buffer_len;
                     chart_transmission_successes++;
                     chart_receptions++;
@@ -1214,7 +1214,7 @@ void *backends_main(void *ptr) {
 
 #if HAVE_MONGOC
     if(do_mongodb) {
-        mongodb_cleanup();
+        backends_mongodb_cleanup();
         freez(mongodb_uri);
         freez(mongodb_database);
         freez(mongodb_collection);

backends/mongodb/Makefile.am

@@ -6,7 +6,3 @@ MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
 dist_noinst_DATA = \
     README.md \
     $(NULL)
-
-dist_libconfig_DATA = \
-    mongodb.conf \
-    $(NULL)

backends/mongodb/mongodb.c

@@ -6,10 +6,10 @@
 
 #define CONFIG_FILE_LINE_MAX ((CONFIG_MAX_NAME + CONFIG_MAX_VALUE + 1024) * 2)
 
-mongoc_client_t *mongodb_client;
-mongoc_collection_t *mongodb_collection;
+static mongoc_client_t *mongodb_client;
+static mongoc_collection_t *mongodb_collection;
 
-int mongodb_init(const char *uri_string,
+int backends_mongodb_init(const char *uri_string,
                  const char *database_string,
                  const char *collection_string,
                  int32_t default_socket_timeout) {
@@ -47,7 +47,7 @@ int mongodb_init(const char *uri_string,
     return 0;
 }
 
-void free_bson(bson_t **insert, size_t n_documents) {
+void backends_free_bson(bson_t **insert, size_t n_documents) {
     size_t i;
 
     for(i = 0; i < n_documents; i++)
@@ -56,7 +56,7 @@ void free_bson(bson_t **insert, size_t n_documents) {
     free(insert);
 }
 
-int mongodb_insert(char *data, size_t n_metrics) {
+int backends_mongodb_insert(char *data, size_t n_metrics) {
     bson_t **insert = calloc(n_metrics, sizeof(bson_t *));
     bson_error_t error;
     char *start = data, *end = data;
@@ -77,7 +77,7 @@ int mongodb_insert(char *data, size_t n_metrics) {
 
         if(unlikely(!insert[n_documents])) {
            error("BACKEND: %s", error.message);
-           free_bson(insert, n_documents);
+           backends_free_bson(insert, n_documents);
            return 1;
         }
 
@@ -88,16 +88,16 @@ int mongodb_insert(char *data, size_t n_metrics) {
 
     if(unlikely(!mongoc_collection_insert_many(mongodb_collection, (const bson_t **)insert, n_documents, NULL, NULL, &error))) {
        error("BACKEND: %s", error.message);
-       free_bson(insert, n_documents);
+       backends_free_bson(insert, n_documents);
        return 1;
     }
 
-    free_bson(insert, n_documents);
+    backends_free_bson(insert, n_documents);
 
     return 0;
 }
 
-void mongodb_cleanup() {
+void backends_mongodb_cleanup() {
     mongoc_collection_destroy(mongodb_collection);
     mongoc_client_destroy(mongodb_client);
     mongoc_cleanup();

backends/mongodb/mongodb.h

@@ -5,11 +5,11 @@
 
 #include "backends/backends.h"
 
-extern int mongodb_init(const char *uri_string, const char *database_string, const char *collection_string, const int32_t socket_timeout);
+extern int backends_mongodb_init(const char *uri_string, const char *database_string, const char *collection_string, const int32_t socket_timeout);
 
-extern int mongodb_insert(char *data, size_t n_metrics);
+extern int backends_mongodb_insert(char *data, size_t n_metrics);
 
-extern void mongodb_cleanup();
+extern void backends_mongodb_cleanup();
 
 extern int read_mongodb_conf(const char *path, char **uri_p, char **database_p, char **collection_p);
 

configure.ac

@@ -1334,6 +1334,7 @@ AC_CONFIG_FILES([
     exporting/prometheus/Makefile
     exporting/prometheus/remote_write/Makefile
     exporting/aws_kinesis/Makefile
+    exporting/mongodb/Makefile
     exporting/tests/Makefile
     health/Makefile
     health/notifications/Makefile

exporting/Makefile.am

@@ -10,6 +10,7 @@ SUBDIRS = \
     opentsdb \
     prometheus \
     aws_kinesis \
+    mongodb \
     $(NULL)
 
 dist_noinst_DATA = \

exporting/exporting_engine.h

@@ -75,6 +75,11 @@ struct aws_kinesis_specific_config {
     char *secure_key;
 };
 
+struct mongodb_specific_config {
+    char *database;
+    char *collection;
+};
+
 struct engine_config {
     const char *prefix;
     const char *hostname;
@@ -140,6 +145,7 @@ struct engine {
     time_t now;
 
     int aws_sdk_initialized;
+    int mongoc_initialized;
 
     struct instance *instance_root;
 };

exporting/init_connectors.c

@@ -4,8 +4,18 @@
 #include "graphite/graphite.h"
 #include "json/json.h"
 #include "opentsdb/opentsdb.h"
-#include "aws_kinesis/aws_kinesis.h"
+
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
 #include "prometheus/remote_write/remote_write.h"
+#endif
+
+#if HAVE_KINESIS
+#include "aws_kinesis/aws_kinesis.h"
+#endif
+
+#if HAVE_MONGOC
+#include "mongodb/mongodb.h"
+#endif
 
 /**
  * Initialize connectors
@@ -48,6 +58,12 @@ int init_connectors(struct engine *engine)
 #if HAVE_KINESIS
                 if (init_aws_kinesis_instance(instance) != 0)
                     return 1;
+#endif
+                break;
+            case BACKEND_TYPE_MONGODB:
+#if HAVE_MONGOC
+                if (init_mongodb_instance(instance) != 0)
+                    return 1;
 #endif
                 break;
             default:

exporting/mongodb/Makefile.am

@@ -0,0 +1,8 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+AUTOMAKE_OPTIONS = subdir-objects
+MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
+
+dist_noinst_DATA = \
+    README.md \
+    $(NULL)

exporting/mongodb/README.md

@@ -0,0 +1,31 @@
+# MongoDB exporting connector
+
+You can use the MongoDB connector and the experimental [exporting engine](../README.md) to archive your agent's metrics
+to a MongoDB database for long-term storage, further analysis, or correlation with data from other sources.
+
+## Prerequisites
+
+To use MongoDB as an external storage for long-term archiving, you should first
+[install](http://mongoc.org/libmongoc/current/installing.html) `libmongoc` 1.7.0 or higher. Next, re-install Netdata
+from the source, which detects that the required library is now available.
+
+## Configuration
+
+To enable data exporting to a MongoDB database, run `./edit-config exporting.conf`
+in the Netdata configuration directory and set the following options:
+
+```conf
+[mongodb:my_instance]
+    enabled = yes
+    destination = mongodb://<hostname>
+    database = your_database_name
+    collection = your_collection_name
+```
+
+You can find more information about the `destination` string URI format in the MongoDB
+[documentation](https://docs.mongodb.com/manual/reference/connection-string/)
+
+The default socket timeout depends on the exporting connector update interval. The timeout is 500 ms shorter than the
+interval (but not less than 1000 ms). You can alter the timeout using the `sockettimeoutms` MongoDB URI option.
+
+[![analytics](https://www.google-analytics.com/collect?v=1&aip=1&t=pageview&_s=1&ds=github&dr=https%3A%2F%2Fgithub.com%2Fnetdata%2Fnetdata&dl=https%3A%2F%2Fmy-netdata.io%2Fgithub%2Fexporting%2Fmongodb%2FREADME&_u=MAC~&cid=5792dfd7-8dc4-476b-af31-da2fdb9f93d2&tid=UA-64295674-3)](<>)