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

1 files changed, 27 insertions, 3 deletions

diff --git a/exporting/read_config.c b/exporting/read_config.c
index 122ceef79d..3602cc57f4 100644
--- a/exporting/read_config.c
+++ b/exporting/read_config.c
@@ -145,6 +145,8 @@ EXPORTING_CONNECTOR_TYPE exporting_select_type(const char *type)
         return EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE;
     } else if (!strcmp(type, "kinesis") || !strcmp(type, "kinesis:plaintext")) {
         return EXPORTING_CONNECTOR_TYPE_KINESIS;
+    } else if (!strcmp(type, "pubsub") || !strcmp(type, "pubsub:plaintext")) {
+        return EXPORTING_CONNECTOR_TYPE_PUBSUB;
     } else if (!strcmp(type, "mongodb") || !strcmp(type, "mongodb:plaintext"))
         return EXPORTING_CONNECTOR_TYPE_MONGODB;
 
@@ -288,6 +290,7 @@ struct engine *read_exporting_config()
     while (tmp_ci_list) {
         struct instance *tmp_instance;
         char *instance_name;
+        char *default_destination = "localhost";
 
         info("Instance %s on %s", tmp_ci_list->local_ci.instance_name, tmp_ci_list->local_ci.connector_name);
 
@@ -310,6 +313,13 @@ struct engine *read_exporting_config()
         }
 #endif
 
+#ifndef ENABLE_EXPORTING_PUBSUB
+        if (tmp_ci_list->backend_type == EXPORTING_CONNECTOR_TYPE_PUBSUB) {
+            error("Google Cloud Pub/Sub support isn't compiled");
+            goto next_connector_instance;
+        }
+#endif
+
 #ifndef HAVE_MONGOC
         if (tmp_ci_list->backend_type == EXPORTING_CONNECTOR_TYPE_MONGODB) {
             error("MongoDB support isn't compiled");
@@ -328,7 +338,6 @@ struct engine *read_exporting_config()
 
         tmp_instance->config.name = strdupz(tmp_ci_list->local_ci.instance_name);
 
-        tmp_instance->config.destination = strdupz(exporter_get(instance_name, "destination", "localhost"));
 
         tmp_instance->config.update_every =
             exporter_get_number(instance_name, EXPORTING_UPDATE_EVERY_OPTION_NAME, EXPORTING_UPDATE_EVERY_DEFAULT);
@@ -376,15 +385,28 @@ struct engine *read_exporting_config()
             struct aws_kinesis_specific_config *connector_specific_config =
                 callocz(1, sizeof(struct aws_kinesis_specific_config));
 
+            default_destination = "us-east-1";
+
             tmp_instance->config.connector_specific_config = connector_specific_config;
 
             connector_specific_config->stream_name = strdupz(exporter_get(instance_name, "stream name", "netdata"));
-
             connector_specific_config->auth_key_id = strdupz(exporter_get(instance_name, "aws_access_key_id", ""));
-
             connector_specific_config->secure_key = strdupz(exporter_get(instance_name, "aws_secret_access_key", ""));
         }
 
+        if (tmp_instance->config.type == EXPORTING_CONNECTOR_TYPE_PUBSUB) {
+            struct pubsub_specific_config *connector_specific_config =
+                callocz(1, sizeof(struct pubsub_specific_config));
+
+            default_destination = "pubsub.googleapis.com";
+
+            tmp_instance->config.connector_specific_config = connector_specific_config;
+
+            connector_specific_config->credentials_file = strdupz(exporter_get(instance_name, "credentials file", ""));
+            connector_specific_config->project_id = strdupz(exporter_get(instance_name, "project id", ""));
+            connector_specific_config->topic_id = strdupz(exporter_get(instance_name, "topic id", ""));
+        }
+
         if (tmp_instance->config.type == EXPORTING_CONNECTOR_TYPE_MONGODB) {
             struct mongodb_specific_config *connector_specific_config =
                 callocz(1, sizeof(struct mongodb_specific_config));
@@ -398,6 +420,8 @@ struct engine *read_exporting_config()
                 instance_name, "collection", ""));
         }
 
+        tmp_instance->config.destination = strdupz(exporter_get(instance_name, "destination", default_destination));
+
 #ifdef NETDATA_INTERNAL_CHECKS
         info(
             "     Dest=[%s], upd=[%d], buffer=[%d] timeout=[%ld] options=[%u]",