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

3 files changed, 298 insertions, 1 deletions

diff --git a/exporting/tests/exporting_doubles.c b/exporting/tests/exporting_doubles.c
index 49946ff938..0d3158d46f 100644
--- a/exporting/tests/exporting_doubles.c
+++ b/exporting/tests/exporting_doubles.c
@@ -244,4 +244,86 @@ int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, siz
     check_expected_ptr(lost_bytes);
     return mock_type(int);
 }
-#endif //HAVE_KINESIS
+#endif // HAVE_KINESIS
+
+#if HAVE_MONGOC
+void __wrap_mongoc_init()
+{
+    function_called();
+}
+
+mongoc_uri_t * __wrap_mongoc_uri_new_with_error (const char *uri_string, bson_error_t *error)
+{
+    function_called();
+    check_expected_ptr(uri_string);
+    check_expected_ptr(error);
+    return mock_ptr_type(mongoc_uri_t *);
+}
+
+int32_t __wrap_mongoc_uri_get_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t fallback)
+{
+    function_called();
+    check_expected_ptr(uri);
+    check_expected_ptr(option);
+    check_expected(fallback);
+    return mock_type(int32_t);
+}
+
+bool __wrap_mongoc_uri_set_option_as_int32 (const mongoc_uri_t *uri, const char *option, int32_t value)
+{
+    function_called();
+    check_expected_ptr(uri);
+    check_expected_ptr(option);
+    check_expected(value);
+    return mock_type(bool);
+}
+
+mongoc_client_t * __wrap_mongoc_client_new_from_uri (const mongoc_uri_t *uri)
+{
+    function_called();
+    check_expected_ptr(uri);
+    return mock_ptr_type(mongoc_client_t *);
+}
+
+bool __wrap_mongoc_client_set_appname (mongoc_client_t *client, const char *appname)
+{
+    function_called();
+    check_expected_ptr(client);
+    check_expected_ptr(appname);
+    return mock_type(bool);
+}
+
+mongoc_collection_t *
+__wrap_mongoc_client_get_collection(mongoc_client_t *client, const char *db, const char *collection)
+{
+    function_called();
+    check_expected_ptr(client);
+    check_expected_ptr(db);
+    check_expected_ptr(collection);
+    return mock_ptr_type(mongoc_collection_t *);
+}
+
+void __wrap_mongoc_uri_destroy (mongoc_uri_t *uri)
+{
+    function_called();
+    check_expected_ptr(uri);
+}
+
+bool __wrap_mongoc_collection_insert_many(
+    mongoc_collection_t *collection,
+    const bson_t **documents,
+    size_t n_documents,
+    const bson_t *opts,
+    bson_t *reply,
+    bson_error_t *error)
+{
+    function_called();
+    check_expected_ptr(collection);
+    check_expected_ptr(documents);
+    check_expected(n_documents);
+    check_expected_ptr(opts);
+    check_expected_ptr(reply);
+    check_expected_ptr(error);
+    return mock_type(bool);
+}
+#endif // HAVE_MONGOC
diff --git a/exporting/tests/test_exporting_engine.c b/exporting/tests/test_exporting_engine.c
index 051390dd86..06d8878af5 100644
--- a/exporting/tests/test_exporting_engine.c
+++ b/exporting/tests/test_exporting_engine.c
@@ -977,6 +977,7 @@ static void test_init_aws_kinesis_instance(void **state)
     expect_string(__wrap_kinesis_init, access_key_id, "test_auth_key_id");
     expect_string(__wrap_kinesis_init, secret_key, "test_secure_key");
     expect_value(__wrap_kinesis_init, timeout, 10000);
+
     assert_int_equal(init_aws_kinesis_instance(instance), 0);
 
     assert_ptr_equal(instance->worker, aws_kinesis_connector_worker);
@@ -1066,6 +1067,178 @@ static void test_aws_kinesis_connector_worker(void **state)
 }
 #endif // HAVE_KINESIS
 
+#if HAVE_MONGOC
+static void test_init_mongodb_instance(void **state)
+{
+    struct engine *engine = *state;
+    struct instance *instance = engine->instance_root;
+
+    instance->config.options = EXPORTING_SOURCE_DATA_AS_COLLECTED | EXPORTING_OPTION_SEND_NAMES;
+
+    struct mongodb_specific_config *connector_specific_config = callocz(1, sizeof(struct mongodb_specific_config));
+    instance->config.connector_specific_config = connector_specific_config;
+    connector_specific_config->database = strdupz("test_database");
+    connector_specific_config->collection = strdupz("test_collection");
+    instance->config.buffer_on_failures = 10;
+
+    expect_function_call(__wrap_mongoc_init);
+    expect_function_call(__wrap_mongoc_uri_new_with_error);
+    expect_string(__wrap_mongoc_uri_new_with_error, uri_string, "localhost");
+    expect_not_value(__wrap_mongoc_uri_new_with_error, error, NULL);
+    will_return(__wrap_mongoc_uri_new_with_error, 0xf1);
+
+    expect_function_call(__wrap_mongoc_uri_get_option_as_int32);
+    expect_value(__wrap_mongoc_uri_get_option_as_int32, uri, 0xf1);
+    expect_string(__wrap_mongoc_uri_get_option_as_int32, option, MONGOC_URI_SOCKETTIMEOUTMS);
+    expect_value(__wrap_mongoc_uri_get_option_as_int32, fallback, 1000);
+    will_return(__wrap_mongoc_uri_get_option_as_int32, 1000);
+
+    expect_function_call(__wrap_mongoc_uri_set_option_as_int32);
+    expect_value(__wrap_mongoc_uri_set_option_as_int32, uri, 0xf1);
+    expect_string(__wrap_mongoc_uri_set_option_as_int32, option, MONGOC_URI_SOCKETTIMEOUTMS);
+    expect_value(__wrap_mongoc_uri_set_option_as_int32, value, 1000);
+    will_return(__wrap_mongoc_uri_set_option_as_int32, true);
+
+    expect_function_call(__wrap_mongoc_client_new_from_uri);
+    expect_value(__wrap_mongoc_client_new_from_uri, uri, 0xf1);
+    will_return(__wrap_mongoc_client_new_from_uri, 0xf2);
+
+    expect_function_call(__wrap_mongoc_client_set_appname);
+    expect_value(__wrap_mongoc_client_set_appname, client, 0xf2);
+    expect_string(__wrap_mongoc_client_set_appname, appname, "netdata");
+    will_return(__wrap_mongoc_client_set_appname, true);
+
+    expect_function_call(__wrap_mongoc_client_get_collection);
+    expect_value(__wrap_mongoc_client_get_collection, client, 0xf2);
+    expect_string(__wrap_mongoc_client_get_collection, db, "test_database");
+    expect_string(__wrap_mongoc_client_get_collection, collection, "test_collection");
+    will_return(__wrap_mongoc_client_get_collection, 0xf3);
+
+    expect_function_call(__wrap_mongoc_uri_destroy);
+    expect_value(__wrap_mongoc_uri_destroy, uri, 0xf1);
+
+    assert_int_equal(init_mongodb_instance(instance), 0);
+
+    assert_ptr_equal(instance->worker, mongodb_connector_worker);
+    assert_ptr_equal(instance->start_batch_formatting, NULL);
+    assert_ptr_equal(instance->start_host_formatting, format_host_labels_json_plaintext);
+    assert_ptr_equal(instance->start_chart_formatting, NULL);
+    assert_ptr_equal(instance->metric_formatting, format_dimension_collected_json_plaintext);
+    assert_ptr_equal(instance->end_chart_formatting, NULL);
+    assert_ptr_equal(instance->end_host_formatting, flush_host_labels);
+    assert_ptr_equal(instance->end_batch_formatting, format_batch_mongodb);
+    assert_ptr_equal(instance->send_header, NULL);
+    assert_ptr_equal(instance->check_response, NULL);
+
+    assert_ptr_not_equal(instance->buffer, NULL);
+    buffer_free(instance->buffer);
+
+    assert_ptr_not_equal(instance->connector_specific_data, NULL);
+
+    struct mongodb_specific_data *connector_specific_data =
+        (struct mongodb_specific_data *)instance->connector_specific_data;
+    size_t number_of_buffers = 1;
+    struct bson_buffer *current_buffer = connector_specific_data->first_buffer;
+    while (current_buffer->next != connector_specific_data->first_buffer) {
+        current_buffer = current_buffer->next;
+        number_of_buffers++;
+        if (number_of_buffers == (size_t)(instance->config.buffer_on_failures + 1)) {
+            number_of_buffers = 0;
+            break;
+        }
+    }
+    assert_int_equal(number_of_buffers, 9);
+
+    free(connector_specific_config->database);
+    free(connector_specific_config->collection);
+}
+
+static void test_format_batch_mongodb(void **state)
+{
+    struct engine *engine = *state;
+    struct instance *instance = engine->instance_root;
+    struct stats *stats = &instance->stats;
+
+    struct mongodb_specific_data *connector_specific_data = mallocz(sizeof(struct mongodb_specific_data));
+    instance->connector_specific_data = (void *)connector_specific_data;
+
+    struct bson_buffer *current_buffer = callocz(1, sizeof(struct bson_buffer));
+    connector_specific_data->first_buffer = current_buffer;
+    connector_specific_data->first_buffer->next = current_buffer;
+    connector_specific_data->last_buffer = current_buffer;
+
+    BUFFER *buffer = buffer_create(0);
+    buffer_sprintf(buffer, "{ \"metric\": \"test_metric\" }\n");
+    instance->buffer = buffer;
+    stats->chart_buffered_metrics = 1;
+
+    assert_int_equal(format_batch_mongodb(instance), 0);
+
+    assert_int_equal(connector_specific_data->last_buffer->documents_inserted, 1);
+    assert_int_equal(buffer_strlen(buffer), 0);
+
+    size_t len;
+    char *str = bson_as_canonical_extended_json(connector_specific_data->last_buffer->insert[0], &len);
+    assert_string_equal(str, "{ \"metric\" : \"test_metric\" }");
+
+    freez(str);
+    buffer_free(buffer);
+}
+
+static void test_mongodb_connector_worker(void **state)
+{
+    struct engine *engine = *state;
+    struct instance *instance = engine->instance_root;
+
+    struct mongodb_specific_config *connector_specific_config = callocz(1, sizeof(struct mongodb_specific_config));
+    instance->config.connector_specific_config = connector_specific_config;
+    connector_specific_config->database = strdupz("test_database");
+
+    struct mongodb_specific_data *connector_specific_data = callocz(1, sizeof(struct mongodb_specific_data));
+    instance->connector_specific_data = (void *)connector_specific_data;
+    connector_specific_config->collection = strdupz("test_collection");
+
+    struct bson_buffer *buffer = callocz(1, sizeof(struct bson_buffer));
+    buffer->documents_inserted = 1;
+    connector_specific_data->first_buffer = buffer;
+    connector_specific_data->first_buffer->next = buffer;
+
+    connector_specific_data->first_buffer->insert = callocz(1, sizeof(bson_t *));
+    bson_error_t bson_error;
+    connector_specific_data->first_buffer->insert[0] =
+        bson_new_from_json((const uint8_t *)"{ \"test_key\" : \"test_value\" }", -1, &bson_error);
+
+    connector_specific_data->client = mongoc_client_new("mongodb://localhost");
+    connector_specific_data->collection =
+        __real_mongoc_client_get_collection(connector_specific_data->client, "test_database", "test_collection");
+
+    expect_function_call(__wrap_mongoc_collection_insert_many);
+    expect_value(__wrap_mongoc_collection_insert_many, collection, connector_specific_data->collection);
+    expect_value(__wrap_mongoc_collection_insert_many, documents, connector_specific_data->first_buffer->insert);
+    expect_value(__wrap_mongoc_collection_insert_many, n_documents, 1);
+    expect_value(__wrap_mongoc_collection_insert_many, opts, NULL);
+    expect_value(__wrap_mongoc_collection_insert_many, reply, NULL);
+    expect_not_value(__wrap_mongoc_collection_insert_many, error, NULL);
+    will_return(__wrap_mongoc_collection_insert_many, true);
+
+    mongodb_connector_worker(instance);
+
+    assert_ptr_equal(connector_specific_data->first_buffer->insert, NULL);
+    assert_int_equal(connector_specific_data->first_buffer->documents_inserted, 0);
+    assert_ptr_equal(connector_specific_data->first_buffer, connector_specific_data->first_buffer->next);
+
+    struct stats *stats = &instance->stats;
+    assert_int_equal(stats->chart_sent_bytes, 60);
+    assert_int_equal(stats->chart_transmission_successes, 1);
+    assert_int_equal(stats->chart_receptions, 1);
+    assert_int_equal(stats->chart_sent_bytes, 60);
+    assert_int_equal(stats->chart_sent_metrics, 0);
+
+    free(connector_specific_config->database);
+    free(connector_specific_config->collection);
+}
+#endif // HAVE_MONGOC
+
 int main(void)
 {
     const struct CMUnitTest tests[] = {
@@ -1168,5 +1341,18 @@ int main(void)
     test_res += cmocka_run_group_tests_name("kinesis_exporting_connector", kinesis_tests, NULL, NULL);
 #endif
 
+#if HAVE_MONGOC
+    const struct CMUnitTest mongodb_tests[] = {
+        cmocka_unit_test_setup_teardown(
+            test_init_mongodb_instance, setup_configured_engine, teardown_configured_engine),
+        cmocka_unit_test_setup_teardown(
+            test_format_batch_mongodb, setup_configured_engine, teardown_configured_engine),
+        cmocka_unit_test_setup_teardown(
+            test_mongodb_connector_worker, setup_configured_engine, teardown_configured_engine),
+    };
+
+    test_res += cmocka_run_group_tests_name("mongodb_exporting_connector", mongodb_tests, NULL, NULL);
+#endif
+
     return test_res;
 }
diff --git a/exporting/tests/test_exporting_engine.h b/exporting/tests/test_exporting_engine.h
index 70c5736f16..519e7ae17d 100644
--- a/exporting/tests/test_exporting_engine.h
+++ b/exporting/tests/test_exporting_engine.h
@@ -9,8 +9,18 @@
 #include "exporting/graphite/graphite.h"
 #include "exporting/json/json.h"
 #include "exporting/opentsdb/opentsdb.h"
+
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
 #include "exporting/prometheus/remote_write/remote_write.h"
+#endif
+
+#if HAVE_KINESIS
 #include "exporting/aws_kinesis/aws_kinesis.h"
+#endif
+
+#if HAVE_MONGOC
+#include "exporting/mongodb/mongodb.h"
+#endif
 
 #include <stdarg.h>
 #include <stddef.h>
@@ -128,6 +138,25 @@ void __wrap_kinesis_put_record(
     size_t data_len);
 int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, size_t *sent_bytes, size_t *lost_bytes);
 
+void __wrap_mongoc_init();
+mongoc_uri_t *__wrap_mongoc_uri_new_with_error(const char *uri_string, bson_error_t *error);
+int32_t __wrap_mongoc_uri_get_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t fallback);
+bool __wrap_mongoc_uri_set_option_as_int32(const mongoc_uri_t *uri, const char *option, int32_t value);
+mongoc_client_t *__wrap_mongoc_client_new_from_uri(const mongoc_uri_t *uri);
+bool __wrap_mongoc_client_set_appname(mongoc_client_t *client, const char *appname);
+mongoc_collection_t *
+__wrap_mongoc_client_get_collection(mongoc_client_t *client, const char *db, const char *collection);
+mongoc_collection_t *
+__real_mongoc_client_get_collection(mongoc_client_t *client, const char *db, const char *collection);
+void __wrap_mongoc_uri_destroy(mongoc_uri_t *uri);
+bool __wrap_mongoc_collection_insert_many(
+    mongoc_collection_t *collection,
+    const bson_t **documents,
+    size_t n_documents,
+    const bson_t *opts,
+    bson_t *reply,
+    bson_error_t *error);
+
 // -----------------------------------------------------------------------
 // fixtures