Viewing file:  aws_kinesis.c (7.33 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
// SPDX-License-Identifier: GPL-3.0-or-later
#include "aws_kinesis.h"
/**
* Clean AWS Kinesis *
*/
void aws_kinesis_cleanup(struct instance *instance)
{
info("EXPORTING: cleaning up instance %s ...", instance->config.name);
kinesis_shutdown(instance->connector_specific_data);
freez(instance->connector_specific_data);
struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
if (connector_specific_config) {
freez(connector_specific_config->auth_key_id);
freez(connector_specific_config->secure_key);
freez(connector_specific_config->stream_name);
freez(connector_specific_config);
}
info("EXPORTING: instance %s exited", instance->config.name);
instance->exited = 1;
}
/**
* Initialize AWS Kinesis connector instance
*
* @param instance an instance data structure.
* @return Returns 0 on success, 1 on failure.
*/
int init_aws_kinesis_instance(struct instance *instance)
{
instance->worker = aws_kinesis_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 AWS Kinesis exporting connector instance %s", instance->config.name);
return 1;
}
if (uv_mutex_init(&instance->mutex))
return 1;
if (uv_cond_init(&instance->cond_var))
return 1;
if (!instance->engine->aws_sdk_initialized) {
aws_sdk_init();
instance->engine->aws_sdk_initialized = 1;
}
struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
struct aws_kinesis_specific_data *connector_specific_data = callocz(1, sizeof(struct aws_kinesis_specific_data));
instance->connector_specific_data = (void *)connector_specific_data;
if (!strcmp(connector_specific_config->stream_name, "")) {
error("stream name is a mandatory Kinesis parameter but it is not configured");
return 1;
}
kinesis_init(
(void *)connector_specific_data,
instance->config.destination,
connector_specific_config->auth_key_id,
connector_specific_config->secure_key,
instance->config.timeoutms);
return 0;
}
/**
* AWS Kinesis connector worker
*
* Runs in a separate thread for every instance.
*
* @param instance_p an instance data structure.
*/
void aws_kinesis_connector_worker(void *instance_p)
{
struct instance *instance = (struct instance *)instance_p;
struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
struct aws_kinesis_specific_data *connector_specific_data = instance->connector_specific_data;
while (!instance->engine->exit) {
unsigned long long partition_key_seq = 0;
struct stats *stats = &instance->stats;
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;
size_t sent = 0;
while (sent < buffer_len) {
char partition_key[KINESIS_PARTITION_KEY_MAX + 1];
snprintf(partition_key, KINESIS_PARTITION_KEY_MAX, "netdata_%llu", partition_key_seq++);
size_t partition_key_len = strnlen(partition_key, KINESIS_PARTITION_KEY_MAX);
const char *first_char = buffer_tostring(buffer) + sent;
size_t record_len = 0;
// split buffer into chunks of maximum allowed size
if (buffer_len - sent < KINESIS_RECORD_MAX - partition_key_len) {
record_len = buffer_len - sent;
} else {
record_len = KINESIS_RECORD_MAX - partition_key_len;
while (record_len && *(first_char + record_len - 1) != '\n')
record_len--;
}
char error_message[ERROR_LINE_MAX + 1] = "";
debug(
D_BACKEND,
"EXPORTING: kinesis_put_record(): dest = %s, id = %s, key = %s, stream = %s, partition_key = %s, \
buffer = %zu, record = %zu",
instance->config.destination,
connector_specific_config->auth_key_id,
connector_specific_config->secure_key,
connector_specific_config->stream_name,
partition_key,
buffer_len,
record_len);
kinesis_put_record(
connector_specific_data, connector_specific_config->stream_name, partition_key, first_char, record_len);
sent += record_len;
stats->transmission_successes++;
size_t sent_bytes = 0, lost_bytes = 0;
if (unlikely(kinesis_get_result(
connector_specific_data->request_outcomes, error_message, &sent_bytes, &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 database backend '%s'. Willing to write %zu bytes, wrote %zu bytes.",
instance->config.destination, sent_bytes, sent_bytes - lost_bytes);
stats->transmission_failures++;
stats->data_lost_events++;
stats->lost_bytes += lost_bytes;
// estimate the number of lost metrics
stats->lost_metrics += (collected_number)(
stats->buffered_metrics *
(buffer_len && (lost_bytes > buffer_len) ? (double)lost_bytes / buffer_len : 1));
break;
} else {
stats->receptions++;
}
if (unlikely(instance->engine->exit))
break;
}
stats->sent_bytes += sent;
if (likely(sent == buffer_len))
stats->sent_metrics = stats->buffered_metrics;
buffer_flush(buffer);
send_internal_metrics(instance);
stats->buffered_metrics = 0;
uv_mutex_unlock(&instance->mutex);
#ifdef UNIT_TESTING
return;
#endif
}
aws_kinesis_cleanup(instance);
}
|