| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "pubsub.h" |
| 4 | |
| 5 | /** |
| 6 | * Initialize Pub/Sub connector instance |
| 7 | * |
| 8 | * @param instance an instance data structure. |
| 9 | * @return Returns 0 on success, 1 on failure. |
| 10 | */ |
| 11 | int init_pubsub_instance(struct instance *instance) |
| 12 | { |
| 13 | instance->worker = pubsub_connector_worker; |
| 14 | |
| 15 | instance->start_batch_formatting = NULL; |
| 16 | instance->start_host_formatting = format_host_labels_json_plaintext; |
| 17 | instance->start_chart_formatting = NULL; |
| 18 | |
| 19 | |
| 20 | if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AS_COLLECTED) |
| 21 | instance->metric_formatting = format_dimension_collected_json_plaintext; |
| 22 | else |
| 23 | instance->metric_formatting = format_dimension_stored_json_plaintext; |
| 24 | |
| 25 | instance->end_chart_formatting = NULL; |
| 26 | instance->variables_formatting = NULL; |
| 27 | instance->end_host_formatting = flush_host_labels; |
| 28 | instance->end_batch_formatting = NULL; |
| 29 | |
| 30 | instance->prepare_header = NULL; |
| 31 | instance->check_response = NULL; |
| 32 | |
| 33 | instance->buffer = (void *)buffer_create(0, &netdata_buffers_statistics.buffers_exporters); |
| 34 | if (!instance->buffer) { |
| 35 | netdata_log_error("EXPORTING: cannot create buffer for Pub/Sub exporting connector instance %s", instance->config.name); |
| 36 | return 1; |
| 37 | } |
| 38 | netdata_mutex_init(&instance->mutex); |
| 39 | netdata_cond_init(&instance->cond_var); |
| 40 | |
| 41 | struct pubsub_specific_data *connector_specific_data = callocz(1, sizeof(struct pubsub_specific_data)); |
| 42 | instance->connector_specific_data = (void *)connector_specific_data; |
| 43 | |
| 44 | struct pubsub_specific_config *connector_specific_config = |
| 45 | (struct pubsub_specific_config *)instance->config.connector_specific_config; |
| 46 | char error_message[ERROR_LINE_MAX + 1] = ""; |
| 47 | if (pubsub_init( |
| 48 | (void *)connector_specific_data, error_message, instance->config.destination, |
| 49 | connector_specific_config->credentials_file, connector_specific_config->project_id, |
| 50 | connector_specific_config->topic_id)) { |
| 51 | netdata_log_error( |
| 52 | "EXPORTING: Cannot initialize a Pub/Sub publisher for instance %s: %s", |
| 53 | instance->config.name, error_message); |
| 54 | return 1; |
| 55 | } |
| 56 | |
| 57 | return 0; |
| 58 | } |
| 59 | |
| 60 | /** |
| 61 | * Clean a PubSub connector instance |
| 62 | * |
| 63 | * @param instance an instance data structure. |
| 64 | */ |
| 65 | void clean_pubsub_instance(struct instance *instance) |
| 66 | { |
| 67 | netdata_log_info("EXPORTING: cleaning up instance %s ...", instance->config.name); |
| 68 | |
| 69 | struct pubsub_specific_data *connector_specific_data = |
| 70 | (struct pubsub_specific_data *)instance->connector_specific_data; |
| 71 | pubsub_cleanup(connector_specific_data); |
| 72 | freez(connector_specific_data); |
| 73 | |
| 74 | buffer_free(instance->buffer); |
| 75 | |
| 76 | struct pubsub_specific_config *connector_specific_config = |
| 77 | (struct pubsub_specific_config *)instance->config.connector_specific_config; |
| 78 | freez(connector_specific_config->credentials_file); |
| 79 | freez(connector_specific_config->project_id); |
| 80 | freez(connector_specific_config->topic_id); |
| 81 | freez(connector_specific_config); |
| 82 | |
| 83 | netdata_log_info("EXPORTING: instance %s exited", instance->config.name); |
| 84 | instance->exited = 1; |
| 85 | |
| 86 | return; |
| 87 | } |
| 88 | |
| 89 | /** |
| 90 | * Pub/Sub connector worker |
| 91 | * |
| 92 | * Runs in a separate thread for every instance. |
| 93 | * |
| 94 | * @param instance_p an instance data structure. |
| 95 | */ |
| 96 | void *pubsub_connector_worker(void *instance_p) |
| 97 | { |
| 98 | struct instance *instance = (struct instance *)instance_p; |
| 99 | struct pubsub_specific_config *connector_specific_config = instance->config.connector_specific_config; |
| 100 | struct pubsub_specific_data *connector_specific_data = instance->connector_specific_data; |
| 101 | |
| 102 | while (!instance->engine->exit) { |
| 103 | struct stats *stats = &instance->stats; |
| 104 | char error_message[ERROR_LINE_MAX + 1] = ""; |
| 105 | |
| 106 | netdata_mutex_lock(&instance->mutex); |
| 107 | while (!instance->data_is_ready) |
| 108 | netdata_cond_wait(&instance->cond_var, &instance->mutex); |
| 109 | instance->data_is_ready = 0; |
| 110 | |
| 111 | |
| 112 | if (unlikely(instance->engine->exit)) { |
| 113 | netdata_mutex_unlock(&instance->mutex); |
| 114 | break; |
| 115 | } |
| 116 | |
| 117 | // reset the monitoring chart counters |
| 118 | stats->received_bytes = |
| 119 | stats->sent_bytes = |
| 120 | stats->sent_metrics = |
| 121 | stats->lost_metrics = |
| 122 | stats->receptions = |
| 123 | stats->transmission_successes = |
| 124 | stats->transmission_failures = |
| 125 | stats->data_lost_events = |
| 126 | stats->lost_bytes = |
| 127 | stats->reconnects = 0; |
| 128 | |
| 129 | BUFFER *buffer = (BUFFER *)instance->buffer; |
| 130 | size_t buffer_len = buffer_strlen(buffer); |
| 131 | |
| 132 | stats->buffered_bytes = buffer_len; |
| 133 | |
| 134 | if (pubsub_add_message(instance->connector_specific_data, (char *)buffer_tostring(buffer))) { |
| 135 | netdata_log_error("EXPORTING: Instance %s: Cannot add data to a message", instance->config.name); |
| 136 | |
| 137 | stats->data_lost_events++; |
| 138 | stats->lost_metrics += stats->buffered_metrics; |
| 139 | stats->lost_bytes += buffer_len; |
| 140 | |
| 141 | goto cleanup; |
| 142 | } |
| 143 | |
| 144 | netdata_log_debug( |
| 145 | D_EXPORTING, "EXPORTING: pubsub_publish(): project = %s, topic = %s, buffer = %zu", |
| 146 | connector_specific_config->project_id, connector_specific_config->topic_id, buffer_len); |
| 147 | |
| 148 | if (pubsub_publish((void *)connector_specific_data, error_message, stats->buffered_metrics, buffer_len)) { |
| 149 | netdata_log_error("EXPORTING: Instance: %s: Cannot publish a message: %s", instance->config.name, error_message); |
| 150 | |
| 151 | stats->transmission_failures++; |
| 152 | stats->data_lost_events++; |
| 153 | stats->lost_metrics += stats->buffered_metrics; |
| 154 | stats->lost_bytes += buffer_len; |
| 155 | |
| 156 | goto cleanup; |
| 157 | } |
| 158 | |
| 159 | stats->sent_bytes = buffer_len; |
| 160 | stats->transmission_successes++; |
| 161 | |
| 162 | size_t sent_metrics = 0, lost_metrics = 0, sent_bytes = 0, lost_bytes = 0; |
| 163 | |
| 164 | if (unlikely(pubsub_get_result( |
| 165 | connector_specific_data, error_message, &sent_metrics, &sent_bytes, &lost_metrics, &lost_bytes))) { |
| 166 | // oops! we couldn't send (all or some of the) data |
| 167 | netdata_log_error("EXPORTING: %s", error_message); |
| 168 | netdata_log_error( |
| 169 | "EXPORTING: failed to write data to service '%s'. Willing to write %zu bytes, wrote %zu bytes.", |
| 170 | instance->config.destination, lost_bytes, sent_bytes); |
| 171 | |
| 172 | stats->transmission_failures++; |
| 173 | stats->data_lost_events++; |
| 174 | stats->lost_metrics += lost_metrics; |
| 175 | stats->lost_bytes += lost_bytes; |
| 176 | } else { |
| 177 | stats->receptions++; |
| 178 | stats->sent_metrics = sent_metrics; |
| 179 | } |
| 180 | |
| 181 | cleanup: |
| 182 | send_internal_metrics(instance); |
| 183 | |
| 184 | buffer_flush(buffer); |
| 185 | stats->buffered_metrics = 0; |
| 186 | |
| 187 | netdata_mutex_unlock(&instance->mutex); |
| 188 | |
| 189 | #ifdef UNIT_TESTING |
| 190 | break; |
| 191 | #endif |
| 192 | } |
| 193 | |
| 194 | clean_pubsub_instance(instance); |
| 195 | return NULL; |
| 196 | } |