summaryrefslogtreecommitdiffstats
path: root/exporting/pubsub/pubsub.c
diff options
context:
space:
mode:
Diffstat (limited to 'exporting/pubsub/pubsub.c')
-rw-r--r--exporting/pubsub/pubsub.c194
1 files changed, 194 insertions, 0 deletions
diff --git a/exporting/pubsub/pubsub.c b/exporting/pubsub/pubsub.c
new file mode 100644
index 0000000..336a096
--- /dev/null
+++ b/exporting/pubsub/pubsub.c
@@ -0,0 +1,194 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "pubsub.h"
+
+/**
+ * Initialize Pub/Sub connector instance
+ *
+ * @param instance an instance data structure.
+ * @return Returns 0 on success, 1 on failure.
+ */
+int init_pubsub_instance(struct instance *instance)
+{
+ instance->worker = pubsub_connector_worker;
+
+ instance->start_batch_formatting = NULL;
+ instance->start_host_formatting = format_host_labels_json_plaintext;
+ instance->start_chart_formatting = NULL;
+
+
+ if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AS_COLLECTED)
+ instance->metric_formatting = format_dimension_collected_json_plaintext;
+ else
+ instance->metric_formatting = format_dimension_stored_json_plaintext;
+
+ instance->end_chart_formatting = NULL;
+ instance->end_host_formatting = flush_host_labels;
+ instance->end_batch_formatting = NULL;
+
+ instance->prepare_header = NULL;
+ instance->check_response = NULL;
+
+ instance->buffer = (void *)buffer_create(0);
+ if (!instance->buffer) {
+ error("EXPORTING: cannot create buffer for Pub/Sub exporting connector instance %s", instance->config.name);
+ return 1;
+ }
+ uv_mutex_init(&instance->mutex);
+ uv_cond_init(&instance->cond_var);
+
+ struct pubsub_specific_data *connector_specific_data = callocz(1, sizeof(struct pubsub_specific_data));
+ instance->connector_specific_data = (void *)connector_specific_data;
+
+ struct pubsub_specific_config *connector_specific_config =
+ (struct pubsub_specific_config *)instance->config.connector_specific_config;
+ char error_message[ERROR_LINE_MAX + 1] = "";
+ if (pubsub_init(
+ (void *)connector_specific_data, error_message, instance->config.destination,
+ connector_specific_config->credentials_file, connector_specific_config->project_id,
+ connector_specific_config->topic_id)) {
+ error(
+ "EXPORTING: Cannot initialize a Pub/Sub publisher for instance %s: %s",
+ instance->config.name, error_message);
+ return 1;
+ }
+
+ return 0;
+}
+
+/**
+ * Clean a PubSub connector instance
+ *
+ * @param instance an instance data structure.
+ */
+void clean_pubsub_instance(struct instance *instance)
+{
+ info("EXPORTING: cleaning up instance %s ...", instance->config.name);
+
+ struct pubsub_specific_data *connector_specific_data =
+ (struct pubsub_specific_data *)instance->connector_specific_data;
+ pubsub_cleanup(connector_specific_data);
+ freez(connector_specific_data);
+
+ buffer_free(instance->buffer);
+
+ struct pubsub_specific_config *connector_specific_config =
+ (struct pubsub_specific_config *)instance->config.connector_specific_config;
+ freez(connector_specific_config->credentials_file);
+ freez(connector_specific_config->project_id);
+ freez(connector_specific_config->topic_id);
+ freez(connector_specific_config);
+
+ info("EXPORTING: instance %s exited", instance->config.name);
+ instance->exited = 1;
+
+ return;
+}
+
+/**
+ * Pub/Sub connector worker
+ *
+ * Runs in a separate thread for every instance.
+ *
+ * @param instance_p an instance data structure.
+ */
+void pubsub_connector_worker(void *instance_p)
+{
+ struct instance *instance = (struct instance *)instance_p;
+ struct pubsub_specific_config *connector_specific_config = instance->config.connector_specific_config;
+ struct pubsub_specific_data *connector_specific_data = instance->connector_specific_data;
+
+ while (!instance->engine->exit) {
+ struct stats *stats = &instance->stats;
+ char error_message[ERROR_LINE_MAX + 1] = "";
+
+ uv_mutex_lock(&instance->mutex);
+ while (!instance->data_is_ready)
+ uv_cond_wait(&instance->cond_var, &instance->mutex);
+ instance->data_is_ready = 0;
+
+
+ if (unlikely(instance->engine->exit)) {
+ uv_mutex_unlock(&instance->mutex);
+ break;
+ }
+
+ // reset the monitoring chart counters
+ stats->received_bytes =
+ stats->sent_bytes =
+ stats->sent_metrics =
+ stats->lost_metrics =
+ stats->receptions =
+ stats->transmission_successes =
+ stats->transmission_failures =
+ stats->data_lost_events =
+ stats->lost_bytes =
+ stats->reconnects = 0;
+
+ BUFFER *buffer = (BUFFER *)instance->buffer;
+ size_t buffer_len = buffer_strlen(buffer);
+
+ stats->buffered_bytes = buffer_len;
+
+ if (pubsub_add_message(instance->connector_specific_data, (char *)buffer_tostring(buffer))) {
+ error("EXPORTING: Instance %s: Cannot add data to a message", instance->config.name);
+
+ stats->data_lost_events++;
+ stats->lost_metrics += stats->buffered_metrics;
+ stats->lost_bytes += buffer_len;
+
+ goto cleanup;
+ }
+
+ debug(
+ D_BACKEND, "EXPORTING: pubsub_publish(): project = %s, topic = %s, buffer = %zu",
+ connector_specific_config->project_id, connector_specific_config->topic_id, buffer_len);
+
+ if (pubsub_publish((void *)connector_specific_data, error_message, stats->buffered_metrics, buffer_len)) {
+ error("EXPORTING: Instance: %s: Cannot publish a message: %s", instance->config.name, error_message);
+
+ stats->transmission_failures++;
+ stats->data_lost_events++;
+ stats->lost_metrics += stats->buffered_metrics;
+ stats->lost_bytes += buffer_len;
+
+ goto cleanup;
+ }
+
+ stats->sent_bytes = buffer_len;
+ stats->transmission_successes++;
+
+ size_t sent_metrics = 0, lost_metrics = 0, sent_bytes = 0, lost_bytes = 0;
+
+ if (unlikely(pubsub_get_result(
+ connector_specific_data, error_message, &sent_metrics, &sent_bytes, &lost_metrics, &lost_bytes))) {
+ // oops! we couldn't send (all or some of the) data
+ error("EXPORTING: %s", error_message);
+ error(
+ "EXPORTING: failed to write data to service '%s'. Willing to write %zu bytes, wrote %zu bytes.",
+ instance->config.destination, lost_bytes, sent_bytes);
+
+ stats->transmission_failures++;
+ stats->data_lost_events++;
+ stats->lost_metrics += lost_metrics;
+ stats->lost_bytes += lost_bytes;
+ } else {
+ stats->receptions++;
+ stats->sent_metrics = sent_metrics;
+ }
+
+ cleanup:
+ send_internal_metrics(instance);
+
+ buffer_flush(buffer);
+ stats->buffered_metrics = 0;
+
+ uv_mutex_unlock(&instance->mutex);
+
+#ifdef UNIT_TESTING
+ return;
+#endif
+ }
+
+ clean_pubsub_instance(instance);
+}