diff options
author | Vladimir Kobal <vlad@prokk.net> | 2020-05-14 10:54:23 +0300 |
---|---|---|
committer | GitHub <noreply@github.com> | 2020-05-14 10:54:23 +0300 |
commit | b5f8c224a9636c071fc474634380b4a93ea21c28 (patch) | |
tree | 0676b067283d6c1fe46f091a7bba6afc6a362bc3 /exporting/tests | |
parent | b8fce8d15bd9f97e8748305ca8f839a6de889199 (diff) |
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
Diffstat (limited to 'exporting/tests')
-rw-r--r-- | exporting/tests/exporting_doubles.c | 49 | ||||
-rw-r--r-- | exporting/tests/test_exporting_engine.c | 149 | ||||
-rw-r--r-- | exporting/tests/test_exporting_engine.h | 14 |
3 files changed, 212 insertions, 0 deletions
diff --git a/exporting/tests/exporting_doubles.c b/exporting/tests/exporting_doubles.c index 29935a7d19..8fafada929 100644 --- a/exporting/tests/exporting_doubles.c +++ b/exporting/tests/exporting_doubles.c @@ -262,6 +262,55 @@ int __wrap_kinesis_get_result(void *request_outcomes_p, char *error_message, siz } #endif // HAVE_KINESIS +#if ENABLE_EXPORTING_PUBSUB +int __wrap_pubsub_init( + void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file, + const char *project_id, const char *topic_id) +{ + function_called(); + check_expected_ptr(pubsub_specific_data_p); + check_expected_ptr(error_message); + check_expected_ptr(destination); + check_expected_ptr(credentials_file); + check_expected_ptr(project_id); + check_expected_ptr(topic_id); + return mock_type(int); +} + +int __wrap_pubsub_add_message(void *pubsub_specific_data_p, char *data) +{ + function_called(); + check_expected_ptr(pubsub_specific_data_p); + check_expected_ptr(data); + return mock_type(int); +} + +int __wrap_pubsub_publish( + void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes) +{ + function_called(); + check_expected_ptr(pubsub_specific_data_p); + check_expected_ptr(error_message); + check_expected(buffered_metrics); + check_expected(buffered_bytes); + return mock_type(int); +} + +int __wrap_pubsub_get_result( + void *pubsub_specific_data_p, char *error_message, + size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, size_t *lost_bytes) +{ + function_called(); + check_expected_ptr(pubsub_specific_data_p); + check_expected_ptr(error_message); + check_expected_ptr(sent_metrics); + check_expected_ptr(sent_bytes); + check_expected_ptr(lost_metrics); + check_expected_ptr(lost_bytes); + return mock_type(int); +} +#endif // ENABLE_EXPORTING_PUBSUB + #if HAVE_MONGOC void __wrap_mongoc_init() { diff --git a/exporting/tests/test_exporting_engine.c b/exporting/tests/test_exporting_engine.c index 09715ddbc5..3835369356 100644 --- a/exporting/tests/test_exporting_engine.c +++ b/exporting/tests/test_exporting_engine.c @@ -1470,6 +1470,144 @@ static void test_aws_kinesis_connector_worker(void **state) } #endif // HAVE_KINESIS +#if ENABLE_EXPORTING_PUBSUB +static void test_init_pubsub_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 pubsub_specific_config *connector_specific_config = + callocz(1, sizeof(struct pubsub_specific_config)); + instance->config.connector_specific_config = connector_specific_config; + connector_specific_config->credentials_file = strdupz("/test/credentials/file"); + connector_specific_config->project_id = strdupz("test_project_id"); + connector_specific_config->topic_id = strdupz("test_topic_id"); + + expect_function_call(__wrap_pubsub_init); + expect_not_value(__wrap_pubsub_init, pubsub_specific_data_p, NULL); + expect_string(__wrap_pubsub_init, destination, "localhost"); + expect_string(__wrap_pubsub_init, error_message, ""); + expect_string(__wrap_pubsub_init, credentials_file, "/test/credentials/file"); + expect_string(__wrap_pubsub_init, project_id, "test_project_id"); + expect_string(__wrap_pubsub_init, topic_id, "test_topic_id"); + will_return(__wrap_pubsub_init, 0); + + assert_int_equal(init_pubsub_instance(instance), 0); + + assert_ptr_equal(instance->worker, pubsub_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, NULL); + assert_ptr_not_equal(instance->buffer, NULL); + buffer_free(instance->buffer); + assert_ptr_not_equal(instance->connector_specific_data, NULL); + freez(instance->connector_specific_data); + + instance->config.options = EXPORTING_SOURCE_DATA_AVERAGE | EXPORTING_OPTION_SEND_NAMES; + + expect_function_call(__wrap_pubsub_init); + expect_not_value(__wrap_pubsub_init, pubsub_specific_data_p, NULL); + expect_string(__wrap_pubsub_init, destination, "localhost"); + expect_string(__wrap_pubsub_init, error_message, ""); + expect_string(__wrap_pubsub_init, credentials_file, "/test/credentials/file"); + expect_string(__wrap_pubsub_init, project_id, "test_project_id"); + expect_string(__wrap_pubsub_init, topic_id, "test_topic_id"); + will_return(__wrap_pubsub_init, 0); + + assert_int_equal(init_pubsub_instance(instance), 0); + assert_ptr_equal(instance->metric_formatting, format_dimension_stored_json_plaintext); + + free(connector_specific_config->credentials_file); + free(connector_specific_config->project_id); + free(connector_specific_config->topic_id); +} + +static void test_pubsub_connector_worker(void **state) +{ + struct engine *engine = *state; + struct instance *instance = engine->instance_root; + struct stats *stats = &instance->stats; + + __real_mark_scheduled_instances(engine); + + expect_function_call(__wrap_rrdhost_is_exportable); + expect_value(__wrap_rrdhost_is_exportable, instance, instance); + expect_value(__wrap_rrdhost_is_exportable, host, localhost); + will_return(__wrap_rrdhost_is_exportable, 1); + + RRDSET *st = localhost->rrdset_root; + expect_function_call(__wrap_rrdset_is_exportable); + expect_value(__wrap_rrdset_is_exportable, instance, instance); + expect_value(__wrap_rrdset_is_exportable, st, st); + will_return(__wrap_rrdset_is_exportable, 1); + + __real_prepare_buffers(engine); + + struct pubsub_specific_config *connector_specific_config = + callocz(1, sizeof(struct pubsub_specific_config)); + instance->config.connector_specific_config = connector_specific_config; + connector_specific_config->credentials_file = strdupz("/test/credentials/file"); + connector_specific_config->project_id = strdupz("test_project_id"); + connector_specific_config->topic_id = strdupz("test_topic_id"); + + struct pubsub_specific_data *connector_specific_data = callocz(1, sizeof(struct pubsub_specific_data)); + instance->connector_specific_data = (void *)connector_specific_data; + + expect_function_call(__wrap_pubsub_add_message); + expect_not_value(__wrap_pubsub_add_message, pubsub_specific_data_p, NULL); + // The buffer is prepared by Graphite exporting connector + expect_string( + __wrap_pubsub_add_message, data, + "netdata.test-host.chart_name.dimension_name;TAG1=VALUE1 TAG2=VALUE2 123000321 15051\n"); + will_return(__wrap_pubsub_add_message, 0); + + expect_function_call(__wrap_pubsub_publish); + expect_not_value(__wrap_pubsub_publish, pubsub_specific_data_p, NULL); + expect_string(__wrap_pubsub_publish, error_message, ""); + expect_value(__wrap_pubsub_publish, buffered_metrics, 1); + expect_value(__wrap_pubsub_publish, buffered_bytes, 84); + will_return(__wrap_pubsub_publish, 0); + + expect_function_call(__wrap_pubsub_get_result); + expect_not_value(__wrap_pubsub_get_result, pubsub_specific_data_p, NULL); + expect_not_value(__wrap_pubsub_get_result, error_message, NULL); + expect_not_value(__wrap_pubsub_get_result, sent_metrics, NULL); + expect_not_value(__wrap_pubsub_get_result, sent_bytes, NULL); + expect_not_value(__wrap_pubsub_get_result, lost_metrics, NULL); + expect_not_value(__wrap_pubsub_get_result, lost_bytes, NULL); + will_return(__wrap_pubsub_get_result, 0); + + expect_function_call(__wrap_send_internal_metrics); + expect_value(__wrap_send_internal_metrics, instance, instance); + will_return(__wrap_send_internal_metrics, 0); + + pubsub_connector_worker(instance); + + assert_int_equal(stats->buffered_metrics, 0); + assert_int_equal(stats->buffered_bytes, 84); + assert_int_equal(stats->received_bytes, 0); + assert_int_equal(stats->sent_bytes, 84); + assert_int_equal(stats->sent_metrics, 0); + assert_int_equal(stats->lost_metrics, 0); + assert_int_equal(stats->receptions, 1); + assert_int_equal(stats->transmission_successes, 1); + assert_int_equal(stats->transmission_failures, 0); + assert_int_equal(stats->data_lost_events, 0); + assert_int_equal(stats->lost_bytes, 0); + assert_int_equal(stats->reconnects, 0); + + free(connector_specific_config->credentials_file); + free(connector_specific_config->project_id); + free(connector_specific_config->topic_id); +} +#endif // ENABLE_EXPORTING_PUBSUB + #if HAVE_MONGOC static void test_init_mongodb_instance(void **state) { @@ -1776,6 +1914,17 @@ int main(void) test_res += cmocka_run_group_tests_name("kinesis_exporting_connector", kinesis_tests, NULL, NULL); #endif +#if ENABLE_EXPORTING_PUBSUB + const struct CMUnitTest pubsub_tests[] = { + cmocka_unit_test_setup_teardown( + test_init_pubsub_instance, setup_configured_engine, teardown_configured_engine), + cmocka_unit_test_setup_teardown( + test_pubsub_connector_worker, setup_initialized_engine, teardown_initialized_engine), + }; + + test_res += cmocka_run_group_tests_name("pubsub_exporting_connector", pubsub_tests, NULL, NULL); +#endif + #if HAVE_MONGOC const struct CMUnitTest mongodb_tests[] = { cmocka_unit_test_setup_teardown( diff --git a/exporting/tests/test_exporting_engine.h b/exporting/tests/test_exporting_engine.h index 6848eb267c..7df3c39c12 100644 --- a/exporting/tests/test_exporting_engine.h +++ b/exporting/tests/test_exporting_engine.h @@ -18,6 +18,10 @@ #include "exporting/aws_kinesis/aws_kinesis.h" #endif +#if ENABLE_EXPORTING_PUBSUB +#include "exporting/pubsub/pubsub.h" +#endif + #if HAVE_MONGOC #include "exporting/mongodb/mongodb.h" #endif @@ -145,6 +149,16 @@ 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); +int __wrap_pubsub_init( + void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file, + const char *project_id, const char *topic_id); +int __wrap_pubsub_add_message(void *pubsub_specific_data_p, char *data); +int __wrap_pubsub_publish( + void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes); +int __wrap_pubsub_get_result( + void *pubsub_specific_data_p, char *error_message, + size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, 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); |