diff options
Diffstat (limited to 'exporting/prometheus/remote_write')
| -rw-r--r-- | exporting/prometheus/remote_write/Makefile.am | 14 | ||||
| -rw-r--r-- | exporting/prometheus/remote_write/README.md | 41 | ||||
| -rw-r--r-- | exporting/prometheus/remote_write/remote_write.c | 320 | ||||
| -rw-r--r-- | exporting/prometheus/remote_write/remote_write.h | 20 | ||||
| -rw-r--r-- | exporting/prometheus/remote_write/remote_write.proto | 29 | ||||
| -rw-r--r-- | exporting/prometheus/remote_write/remote_write_request.cc | 186 | ||||
| -rw-r--r-- | exporting/prometheus/remote_write/remote_write_request.h | 37 |
7 files changed, 647 insertions, 0 deletions
diff --git a/exporting/prometheus/remote_write/Makefile.am b/exporting/prometheus/remote_write/Makefile.am new file mode 100644 index 0000000000..d049ef48c2 --- /dev/null +++ b/exporting/prometheus/remote_write/Makefile.am @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +AUTOMAKE_OPTIONS = subdir-objects +MAINTAINERCLEANFILES = $(srcdir)/Makefile.in + +CLEANFILES = \ + remote_write.pb.cc \ + remote_write.pb.h \ + $(NULL) + +dist_noinst_DATA = \ + remote_write.proto \ + README.md \ + $(NULL) diff --git a/exporting/prometheus/remote_write/README.md b/exporting/prometheus/remote_write/README.md new file mode 100644 index 0000000000..986f2cafac --- /dev/null +++ b/exporting/prometheus/remote_write/README.md @@ -0,0 +1,41 @@ +# Prometheus remote write exporting connector + +The Prometheus remote write exporting connector uses the exporting engine to send Netdata metrics to your choice of more +than 20 external storage providers for long-term archiving and further analysis. + +## Prerequisites + +To use the Prometheus remote write API with [storage +providers](https://prometheus.io/docs/operating/integrations/#remote-endpoints-and-storage), install +[protobuf](https://developers.google.com/protocol-buffers/) and [snappy](https://github.com/google/snappy) libraries. +Next, re-install Netdata from the source, which detects that the required libraries and +utilities are now available. + +## Configuration + +To enable data exporting to a storage provider using the Prometheus remote write API, run `./edit-config exporting.conf` +in the Netdata configuration directory and set the following options: + +```conf +[remote_write:my_instance] + enabled = yes + destination = example.domain:example_port + remote write URL path = /receive +``` + +`remote write URL path` is used to set an endpoint path for the remote write protocol. The default value is `/receive`. +For example, if your endpoint is `http://example.domain:example_port/storage/read`: + +```conf + destination = example.domain:example_port + remote write URL path = /storage/read +``` + +`buffered` and `lost` dimensions in the Netdata Exporting Connector Data Size operation monitoring chart estimate uncompressed +buffer size on failures. + +## Notes + +The remote write exporting connector does not support `buffer on failures` + +[](<>) diff --git a/exporting/prometheus/remote_write/remote_write.c b/exporting/prometheus/remote_write/remote_write.c new file mode 100644 index 0000000000..248a053c21 --- /dev/null +++ b/exporting/prometheus/remote_write/remote_write.c @@ -0,0 +1,320 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "remote_write.h" + +static int as_collected; +static int homogeneous; +char context[PROMETHEUS_ELEMENT_MAX + 1]; +char chart[PROMETHEUS_ELEMENT_MAX + 1]; +char family[PROMETHEUS_ELEMENT_MAX + 1]; +char units[PROMETHEUS_ELEMENT_MAX + 1] = ""; + +/** + * Send header to a server + * + * @param sock a communication socket. + * @param instance an instance data structure. + * @return Returns 0 on success, 1 on failure. + */ +int prometheus_remote_write_send_header(int *sock, struct instance *instance) +{ + int flags = 0; +#ifdef MSG_NOSIGNAL + flags += MSG_NOSIGNAL; +#endif + + struct prometheus_remote_write_specific_config *connector_specific_config = + instance->config.connector_specific_config; + + static BUFFER *header; + if (!header) + header = buffer_create(0); + + buffer_sprintf( + header, + "POST %s HTTP/1.1\r\n" + "Host: %s\r\n" + "Accept: */*\r\n" + "Content-Length: %zu\r\n" + "Content-Type: application/x-www-form-urlencoded\r\n\r\n", + connector_specific_config->remote_write_path, + instance->engine->config.hostname, + buffer_strlen((BUFFER *)instance->buffer)); + + size_t header_len = buffer_strlen(header); + ssize_t written = send(*sock, buffer_tostring(header), header_len, flags); + + buffer_flush(header); + + if (written != -1 && (size_t)written == header_len) + return 0; + else + return 1; +} + +/** + * Process a responce received after Prometheus remote write connector had sent data + * + * @param buffer a response from a remote service. + * @param instance an instance data structure. + * @return Returns 0 on success, 1 on failure. + */ +int process_prometheus_remote_write_response(BUFFER *buffer, struct instance *instance) +{ + if (unlikely(!buffer)) + return 1; + + const char *s = buffer_tostring(buffer); + int len = buffer_strlen(buffer); + + // do nothing with HTTP responses 200 or 204 + + while (!isspace(*s) && len) { + s++; + len--; + } + s++; + len--; + + if (likely(len > 4 && (!strncmp(s, "200 ", 4) || !strncmp(s, "204 ", 4)))) + return 0; + else + return exporting_discard_response(buffer, instance); +} + +/** + * Initialize Prometheus Remote Write connector instance + * + * @param instance an instance data structure. + * @return Returns 0 on success, 1 on failure. + */ +int init_prometheus_remote_write_instance(struct instance *instance) +{ + instance->worker = simple_connector_worker; + + instance->start_batch_formatting = NULL; + instance->start_host_formatting = format_host_prometheus_remote_write; + instance->start_chart_formatting = format_chart_prometheus_remote_write; + instance->metric_formatting = format_dimension_prometheus_remote_write; + instance->end_chart_formatting = NULL; + instance->end_host_formatting = NULL; + instance->end_batch_formatting = format_batch_prometheus_remote_write; + + instance->send_header = prometheus_remote_write_send_header; + instance->check_response = process_prometheus_remote_write_response; + + 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; + } + uv_mutex_init(&instance->mutex); + uv_cond_init(&instance->cond_var); + + struct prometheus_remote_write_specific_data *connector_specific_data = + callocz(1, sizeof(struct prometheus_remote_write_specific_data)); + instance->connector_specific_data = (void *)connector_specific_data; + + connector_specific_data->write_request = init_write_request(); + + return 0; +} + +/** + * Format host data for Prometheus Remote Write connector + * + * @param instance an instance data structure. + * @param host a data collecting host. + * @return Always returns 0. + */ +int format_host_prometheus_remote_write(struct instance *instance, RRDHOST *host) +{ + struct prometheus_remote_write_specific_data *connector_specific_data = + (struct prometheus_remote_write_specific_data *)instance->connector_specific_data; + + char hostname[PROMETHEUS_ELEMENT_MAX + 1]; + prometheus_label_copy(hostname, instance->engine->config.hostname, PROMETHEUS_ELEMENT_MAX); + + add_host_info( + connector_specific_data->write_request, + "netdata_info", hostname, host->program_name, host->program_version, now_realtime_usec() / USEC_PER_MS); + + if (unlikely(sending_labels_configured(instance))) { + rrdhost_check_rdlock(host); + netdata_rwlock_rdlock(&host->labels_rwlock); + for (struct label *label = host->labels; label; label = label->next) { + if (!should_send_label(instance, label)) + continue; + + char key[PROMETHEUS_ELEMENT_MAX + 1]; + prometheus_name_copy(key, label->key, PROMETHEUS_ELEMENT_MAX); + + char value[PROMETHEUS_ELEMENT_MAX + 1]; + prometheus_label_copy(value, label->value, PROMETHEUS_ELEMENT_MAX); + + add_label(connector_specific_data->write_request, key, value); + } + netdata_rwlock_unlock(&host->labels_rwlock); + } + + return 0; +} + +/** + * Format chart data for Prometheus Remote Write connector + * + * @param instance an instance data structure. + * @param st a chart. + * @return Always returns 0. + */ +int format_chart_prometheus_remote_write(struct instance *instance, RRDSET *st) +{ + prometheus_label_copy( + chart, + (instance->config.options & EXPORTING_OPTION_SEND_NAMES && st->name) ? st->name : st->id, + PROMETHEUS_ELEMENT_MAX); + prometheus_label_copy(family, st->family, PROMETHEUS_ELEMENT_MAX); + prometheus_name_copy(context, st->context, PROMETHEUS_ELEMENT_MAX); + + if (likely(can_send_rrdset(instance, st))) { + as_collected = (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AS_COLLECTED); + homogeneous = 1; + if (as_collected) { + if (rrdset_flag_check(st, RRDSET_FLAG_HOMOGENEOUS_CHECK)) + rrdset_update_heterogeneous_flag(st); + + if (rrdset_flag_check(st, RRDSET_FLAG_HETEROGENEOUS)) + homogeneous = 0; + } else { + if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AVERAGE) + prometheus_units_copy(units, st->units, PROMETHEUS_ELEMENT_MAX, 0); + } + } + + return 0; +} + +/** + * Format dimension data for Prometheus Remote Write connector + * + * @param instance an instance data structure. + * @param rd a dimension. + * @return Always returns 0. + */ +int format_dimension_prometheus_remote_write(struct instance *instance, RRDDIM *rd) +{ + struct prometheus_remote_write_specific_data *connector_specific_data = + (struct prometheus_remote_write_specific_data *)instance->connector_specific_data; + + if (rd->collections_counter && !rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE)) { + char name[PROMETHEUS_LABELS_MAX + 1]; + char dimension[PROMETHEUS_ELEMENT_MAX + 1]; + char *suffix = ""; + + if (as_collected) { + // we need as-collected / raw data + + if (unlikely(rd->last_collected_time.tv_sec < instance->after)) { + debug( + D_BACKEND, + "EXPORTING: not sending dimension '%s' of chart '%s' from host '%s', " + "its last data collection (%lu) is not within our timeframe (%lu to %lu)", + rd->id, rd->rrdset->id, + instance->engine->config.hostname, + (unsigned long)rd->last_collected_time.tv_sec, + (unsigned long)instance->after, + (unsigned long)instance->before); + return 1; + } + + if (homogeneous) { + // all the dimensions of the chart, has the same algorithm, multiplier and divisor + // we add all dimensions as labels + + prometheus_label_copy( + dimension, + (instance->config.options & EXPORTING_OPTION_SEND_NAMES && rd->name) ? rd->name : rd->id, + PROMETHEUS_ELEMENT_MAX); + snprintf(name, PROMETHEUS_LABELS_MAX, "%s_%s%s", instance->engine->config.prefix, context, suffix); + + add_metric( + connector_specific_data->write_request, + name, chart, family, dimension, instance->engine->config.hostname, + rd->last_collected_value, timeval_msec(&rd->last_collected_time)); + } else { + // the dimensions of the chart, do not have the same algorithm, multiplier or divisor + // we create a metric per dimension + + prometheus_name_copy( + dimension, + (instance->config.options & EXPORTING_OPTION_SEND_NAMES && rd->name) ? rd->name : rd->id, + PROMETHEUS_ELEMENT_MAX); + snprintf( + name, PROMETHEUS_LABELS_MAX, "%s_%s_%s%s", instance->engine->config.prefix, context, dimension, + suffix); + + add_metric( + connector_specific_data->write_request, + name, chart, family, NULL, instance->engine->config.hostname, + rd->last_collected_value, timeval_msec(&rd->last_collected_time)); + } + } else { + // we need average or sum of the data + + time_t last_t = instance->before; + calculated_number value = exporting_calculate_value_from_stored_data(instance, rd, &last_t); + + if (!isnan(value) && !isinf(value)) { + if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_AVERAGE) + suffix = "_average"; + else if (EXPORTING_OPTIONS_DATA_SOURCE(instance->config.options) == EXPORTING_SOURCE_DATA_SUM) + suffix = "_sum"; + + prometheus_label_copy( + dimension, + (instance->config.options & EXPORTING_OPTION_SEND_NAMES && rd->name) ? rd->name : rd->id, + PROMETHEUS_ELEMENT_MAX); + snprintf( + name, PROMETHEUS_LABELS_MAX, "%s_%s%s%s", instance->engine->config.prefix, context, units, suffix); + + add_metric( + connector_specific_data->write_request, + name, chart, family, dimension, instance->engine->config.hostname, + value, last_t * MSEC_PER_SEC); + } + } + } + + return 0; +} + +/** + * Format a batch for Prometheus Remote Write connector + * + * @param instance an instance data structure. + * @return Returns 0 on success, 1 on failure. + */ +int format_batch_prometheus_remote_write(struct instance *instance) +{ + struct prometheus_remote_write_specific_data *connector_specific_data = + (struct prometheus_remote_write_specific_data *)instance->connector_specific_data; + + size_t data_size = get_write_request_size(connector_specific_data->write_request); + + if (unlikely(!data_size)) { + error("EXPORTING: write request size is out of range"); + return 1; + } + + BUFFER *buffer = instance->buffer; + + buffer_need_bytes(buffer, data_size); + if (unlikely(pack_and_clear_write_request(connector_specific_data->write_request, buffer->buffer, &data_size))) { + error("EXPORTING: cannot pack write request"); + return 1; + } + buffer->len = data_size; + instance->stats.chart_buffered_bytes = (collected_number)buffer_strlen(buffer); + + return 0; +} diff --git a/exporting/prometheus/remote_write/remote_write.h b/exporting/prometheus/remote_write/remote_write.h new file mode 100644 index 0000000000..050eb0c1a8 --- /dev/null +++ b/exporting/prometheus/remote_write/remote_write.h @@ -0,0 +1,20 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#ifndef NETDATA_EXPORTING_PROMETHEUS_REMOTE_WRITE_H +#define NETDATA_EXPORTING_PROMETHEUS_REMOTE_WRITE_H + +#include "exporting/exporting_engine.h" +#include "exporting/prometheus/prometheus.h" +#include "remote_write_request.h" + +int init_prometheus_remote_write_instance(struct instance *instance); + +int format_host_prometheus_remote_write(struct instance *instance, RRDHOST *host); +int format_chart_prometheus_remote_write(struct instance *instance, RRDSET *st); +int format_dimension_prometheus_remote_write(struct instance *instance, RRDDIM *rd); +int format_batch_prometheus_remote_write(struct instance *instance); + +int prometheus_remote_write_send_header(int *sock, struct instance *instance); +int process_prometheus_remote_write_response(BUFFER *buffer, struct instance *instance); + +#endif //NETDATA_EXPORTING_PROMETHEUS_REMOTE_WRITE_H diff --git a/exporting/prometheus/remote_write/remote_write.proto b/exporting/prometheus/remote_write/remote_write.proto new file mode 100644 index 0000000000..dfde254e13 --- /dev/null +++ b/exporting/prometheus/remote_write/remote_write.proto @@ -0,0 +1,29 @@ +syntax = "proto3"; +package prometheus; + +option cc_enable_arenas = true; + +import "google/protobuf/descriptor.proto"; + +message WriteRequest { + repeated TimeSeries timeseries = 1 [(nullable) = false]; +} + +message TimeSeries { + repeated Label labels = 1 [(nullable) = false]; + repeated Sample samples = 2 [(nullable) = false]; +} + +message Label { + string name = 1; + string value = 2; +} + +message Sample { + double value = 1; + int64 timestamp = 2; +} + +extend google.protobuf.FieldOptions { + bool nullable = 65001; +} diff --git a/exporting/prometheus/remote_write/remote_write_request.cc b/exporting/prometheus/remote_write/remote_write_request.cc new file mode 100644 index 0000000000..48d19efd62 --- /dev/null +++ b/exporting/prometheus/remote_write/remote_write_request.cc @@ -0,0 +1,186 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#include <snappy.h> +#include "remote_write.pb.h" +#include "remote_write_request.h" + +using namespace prometheus; + +google::protobuf::Arena arena; + +/** + * Initialize a write request + * + * @return Returns a new write request + */ +void *init_write_request() +{ + GOOGLE_PROTOBUF_VERIFY_VERSION; + WriteRequest *write_request = google::protobuf::Arena::CreateMessage<WriteRequest>(&arena); + return (void *)write_request; +} + +/** + * Adds information about a host to a write request + * + * @param write_request_p the write request + * @param name the name of a metric which is used for providing the host information + * @param instance the name of the host itself + * @param application the name of a program which sends the information + * @param version the version of the program + * @param timestamp the timestamp for the metric in milliseconds + */ +void add_host_info( + void *write_request_p, + const char *name, const char *instance, const char *application, const char *version, const int64_t timestamp) +{ + WriteRequest *write_request = (WriteRequest *)write_request_p; + TimeSeries *timeseries; + Sample *sample; + Label *label; + + timeseries = write_request->add_timeseries(); + + label = timeseries->add_labels(); + label->set_name("__name__"); + label->set_value(name); + + label = timeseries->add_labels(); + label->set_name("instance"); + label->set_value(instance); + + if (application) { + label = timeseries->add_labels(); + label->set_name("application"); + label->set_value(application); + } + + if (version) { + label = timeseries->add_labels(); + label->set_name("version"); + label->set_value(version); + } + + sample = timeseries->add_samples(); + sample->set_value(1); + sample->set_timestamp(timestamp); +} + +/** + * Adds a label to the last created timeseries + * + * @param write_request_p the write request with the timeseries + * @param key the key of the label + * @param value the value of the label + */ +void add_label(void *write_request_p, char *key, char *value) +{ + WriteRequest *write_request = (WriteRequest *)write_request_p; + TimeSeries *timeseries; + Label *label; + + timeseries = write_request->mutable_timeseries(write_request->timeseries_size() - 1); + + label = timeseries->add_labels(); + label->set_name(key); + label->set_value(value); +} + +/** + * Adds a metric to a write request + * + * @param write_request_p the write request + * @param name the name of the metric + * @param chart the chart, the metric belongs to + * @param family the family, the metric belongs to + * @param dimension the dimension, the metric belongs to + * @param instance the name of the host, the metric belongs to + * @param value the value of the metric + * @param timestamp the timestamp for the metric in milliseconds + */ +void add_metric( + void *write_request_p, + const char *name, const char *chart, const char *family, const char *dimension, const char *instance, + const double value, const int64_t timestamp) +{ + WriteRequest *write_request = (WriteRequest *)write_request_p; + TimeSeries *timeseries; + Sample *sample; + Label *label; + + timeseries = write_request->add_timeseries(); + + label = timeseries->add_labels(); + label->set_name("__name__"); + label->set_value(name); + + label = timeseries->add_labels(); + label->set_name("chart"); + label->set_value(chart); + + label = timeseries->add_labels(); + label->set_name("family"); + label->set_value(family); + + if (dimension) { + label = timeseries->add_labels(); + label->set_name("dimension"); + label->set_value(dimension); + } + + label = timeseries->add_labels(); + label->set_name("instance"); + label->set_value(instance); + + sample = timeseries->add_samples(); + sample->set_value(value); + sample->set_timestamp(timestamp); +} + +/** + * Gets the size of a write request + * + * @param write_request_p the write request + * @return Returns the size of the write request + */ +size_t get_write_request_size(void *write_request_p) +{ + WriteRequest *write_request = (WriteRequest *)write_request_p; + +#if GOOGLE_PROTOBUF_VERSION < 3001000 + size_t size = (size_t)snappy::MaxCompressedLength(write_request->ByteSize()); +#else + size_t size = (size_t)snappy::MaxCompressedLength(write_request->ByteSizeLong()); +#endif + + return (size < INT_MAX) ? size : 0; +} + +/** + * Packs a write request into a buffer and clears the request + * + * @param write_request_p the write request + * @param buffer a buffer, where compressed data is written + * @param size gets the size of the write request, returns the size of the compressed data + * @return Returns 0 on success, 1 on failure + */ +int pack_and_clear_write_request(void *write_request_p, char *buffer, size_t *size) +{ + WriteRequest *write_request = (WriteRequest *)write_request_p; + std::string uncompressed_write_request; + + if (write_request->SerializeToString(&uncompressed_write_request) == false) + return 1; + write_request->clear_timeseries(); + snappy::RawCompress(uncompressed_write_request.data(), uncompressed_write_request.size(), buffer, size); + + return 0; +} + +/** + * Shuts down the Protobuf library + */ +void protocol_buffers_shutdown() +{ + google::protobuf::ShutdownProtobufLibrary(); +} diff --git a/exporting/prometheus/remote_write/remote_write_request.h b/exporting/prometheus/remote_write/remote_write_request.h new file mode 100644 index 0000000000..ea8f0f6795 --- /dev/null +++ b/exporting/prometheus/remote_write/remote_write_request.h @@ -0,0 +1,37 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#ifndef NETDATA_EXPORTING_PROMETHEUS_REMOTE_WRITE_REQUEST_H +#define NETDATA_EXPORTING_PROMETHEUS_REMOTE_WRITE_REQUEST_H + +#ifdef __cplusplus +extern "C" { +#endif + +struct prometheus_remote_write_specific_data { + void *write_request; +}; + +void *init_write_request(); + +void add_host_info( + void *write_request_p, + const char *name, const char *instance, const char *application, const char *version, const int64_t timestamp); + +void add_label(void *write_request_p, char *key, char *value); + +void add_metric( + void *write_request_p, + const char *name, const char *chart, const char *family, const char *dimension, + const char *instance, const double value, const int64_t timestamp); + +size_t get_write_request_size(void *write_request_p); + +int pack_and_clear_write_request(void *write_request_p, char *buffer, size_t *size); + +void protocol_buffers_shutdown(); + +#ifdef __cplusplus +} +#endif + +#endif //NETDATA_EXPORTING_PROMETHEUS_REMOTE_WRITE_REQUEST_H |