Add an AWS Kinesis connector to the exporting engine (#8145)

* Prepare files for the AWS Kinesis exporting connector

* Update the documentation

* Rename functions in backends

* Include the connector to the Netdata buid

* Add initializers and a worker

* Add Kinesis specific configuration options

* Add a compile time configuration check

* Remove the connector data structure

* Restore unit tests

* Fix the compile-time configuration check

* Initialize AWS SDK only once

* Don't create an instance for an unknown exporting connector

* Separate client and request outcome data for every instance

* Fix memory cleanup, document functions

* Add unit tests

* Update the documentation

3 files changed, 16 insertions, 16 deletions

diff --git a/backends/aws_kinesis/aws_kinesis_put_record.cc b/backends/aws_kinesis/aws_kinesis_put_record.cc
index 0c8ece68bf..a8ba4aaca6 100644
--- a/backends/aws_kinesis/aws_kinesis_put_record.cc
+++ b/backends/aws_kinesis/aws_kinesis_put_record.cc
@@ -10,18 +10,18 @@
 
 using namespace Aws;
 
-SDKOptions options;
+static SDKOptions options;
 
-Kinesis::KinesisClient *client;
+static Kinesis::KinesisClient *client;
 
 struct request_outcome {
     Kinesis::Model::PutRecordOutcomeCallable future_outcome;
     size_t data_len;
 };
 
-Vector<request_outcome> request_outcomes;
+static Vector<request_outcome> request_outcomes;
 
-void kinesis_init(const char *region, const char *access_key_id, const char *secret_key, const long timeout) {
+void backends_kinesis_init(const char *region, const char *access_key_id, const char *secret_key, const long timeout) {
     InitAPI(options);
 
     Client::ClientConfiguration config;
@@ -37,13 +37,13 @@ void kinesis_init(const char *region, const char *access_key_id, const char *sec
     }
 }
 
-void kinesis_shutdown() {
+void backends_kinesis_shutdown() {
     Delete(client);
 
     ShutdownAPI(options);
 }
 
-int kinesis_put_record(const char *stream_name, const char *partition_key,
+int backends_kinesis_put_record(const char *stream_name, const char *partition_key,
                        const char *data, size_t data_len) {
     Kinesis::Model::PutRecordRequest request;
 
@@ -56,7 +56,7 @@ int kinesis_put_record(const char *stream_name, const char *partition_key,
     return 0;
 }
 
-int kinesis_get_result(char *error_message, size_t *sent_bytes, size_t *lost_bytes) {
+int backends_kinesis_get_result(char *error_message, size_t *sent_bytes, size_t *lost_bytes) {
     Kinesis::Model::PutRecordOutcome outcome;
     *sent_bytes = 0;
     *lost_bytes = 0;
diff --git a/backends/aws_kinesis/aws_kinesis_put_record.h b/backends/aws_kinesis/aws_kinesis_put_record.h
index f48e420f34..fa3d034591 100644
--- a/backends/aws_kinesis/aws_kinesis_put_record.h
+++ b/backends/aws_kinesis/aws_kinesis_put_record.h
@@ -9,14 +9,14 @@
 extern "C" {
 #endif
 
-void kinesis_init(const char *region, const char *access_key_id, const char *secret_key, const long timeout);
+void backends_kinesis_init(const char *region, const char *access_key_id, const char *secret_key, const long timeout);
 
-void kinesis_shutdown();
+void backends_kinesis_shutdown();
 
-int kinesis_put_record(const char *stream_name, const char *partition_key,
+int backends_kinesis_put_record(const char *stream_name, const char *partition_key,
                        const char *data, size_t data_len);
 
-int kinesis_get_result(char *error_message, size_t *sent_bytes, size_t *lost_bytes);
+int backends_kinesis_get_result(char *error_message, size_t *sent_bytes, size_t *lost_bytes);
 
 #ifdef __cplusplus
 }
diff --git a/backends/backends.c b/backends/backends.c
index a436ca6836..b373b2d05e 100644
--- a/backends/backends.c
+++ b/backends/backends.c
@@ -578,7 +578,7 @@ void *backends_main(void *ptr) {
                 goto cleanup;
             }
 
-            kinesis_init(destination, kinesis_auth_key_id, kinesis_secure_key, timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
+            backends_kinesis_init(destination, kinesis_auth_key_id, kinesis_secure_key, timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
 #else
             error("BACKEND: AWS Kinesis support isn't compiled");
 #endif // HAVE_KINESIS
@@ -860,18 +860,18 @@ void *backends_main(void *ptr) {
 
                 char error_message[ERROR_LINE_MAX + 1] = "";
 
-                debug(D_BACKEND, "BACKEND: kinesis_put_record(): dest = %s, id = %s, key = %s, stream = %s, partition_key = %s, \
+                debug(D_BACKEND, "BACKEND: backends_kinesis_put_record(): dest = %s, id = %s, key = %s, stream = %s, partition_key = %s, \
                       buffer = %zu, record = %zu", destination, kinesis_auth_key_id, kinesis_secure_key, kinesis_stream_name,
                       partition_key, buffer_len, record_len);
 
-                kinesis_put_record(kinesis_stream_name, partition_key, first_char, record_len);
+                backends_kinesis_put_record(kinesis_stream_name, partition_key, first_char, record_len);
 
                 sent += record_len;
                 chart_transmission_successes++;
 
                 size_t sent_bytes = 0, lost_bytes = 0;
 
-                if(unlikely(kinesis_get_result(error_message, &sent_bytes, &lost_bytes))) {
+                if(unlikely(backends_kinesis_get_result(error_message, &sent_bytes, &lost_bytes))) {
                     // oops! we couldn't send (all or some of the) data
                     error("BACKEND: %s", error_message);
                     error("BACKEND: failed to write data to database backend '%s'. Willing to write %zu bytes, wrote %zu bytes.",
@@ -1199,7 +1199,7 @@ void *backends_main(void *ptr) {
 cleanup:
 #if HAVE_KINESIS
     if(do_kinesis) {
-        kinesis_shutdown();
+        backends_kinesis_shutdown();
         freez(kinesis_auth_key_id);
         freez(kinesis_secure_key);
         freez(kinesis_stream_name);