diff options
Diffstat (limited to 'src/fluent-bit/plugins/out_prometheus_remote_write')
5 files changed, 842 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/out_prometheus_remote_write/CMakeLists.txt b/src/fluent-bit/plugins/out_prometheus_remote_write/CMakeLists.txt new file mode 100644 index 000000000..71779cb9d --- /dev/null +++ b/src/fluent-bit/plugins/out_prometheus_remote_write/CMakeLists.txt @@ -0,0 +1,6 @@ +set(src
+ remote_write.c
+ remote_write_conf.c
+ )
+
+FLB_PLUGIN(out_prometheus_remote_write "${src}" "")
diff --git a/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write.c b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write.c new file mode 100644 index 000000000..2349afd61 --- /dev/null +++ b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write.c @@ -0,0 +1,466 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <fluent-bit/flb_output_plugin.h> +#include <fluent-bit/flb_snappy.h> +#include <fluent-bit/flb_gzip.h> +#include <fluent-bit/flb_metrics.h> +#include <fluent-bit/flb_kv.h> + +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS +#include <fluent-bit/flb_aws_credentials.h> +#include <fluent-bit/flb_signv4.h> +#endif +#endif + +#include "remote_write.h" +#include "remote_write_conf.h" + +static int http_post(struct prometheus_remote_write_context *ctx, + const void *body, size_t body_len, + const char *tag, int tag_len) +{ + int ret; + int out_ret = FLB_OK; + size_t b_sent; + void *payload_buf = NULL; + size_t payload_size = 0; + struct flb_upstream *u; + struct flb_connection *u_conn; + struct flb_http_client *c; + struct mk_list *head; + struct flb_config_map_val *mv; + struct flb_slist_entry *key = NULL; + struct flb_slist_entry *val = NULL; + flb_sds_t signature = NULL; + + /* Get upstream context and connection */ + u = ctx->u; + u_conn = flb_upstream_conn_get(u); + if (!u_conn) { + flb_plg_error(ctx->ins, "no upstream connections available to %s:%i", + u->tcp_host, u->tcp_port); + return FLB_RETRY; + } + + /* Map payload */ + + if (strcasecmp(ctx->compression, "snappy") == 0) { + ret = flb_snappy_compress((void *) body, body_len, + (char **) &payload_buf, + &payload_size); + } + else if (strcasecmp(ctx->compression, "gzip") == 0) { + ret = flb_gzip_compress((void *) body, body_len, + &payload_buf, &payload_size); + } + else { + payload_buf = (void *) body; + payload_size = body_len; + + ret = 0; + } + + if (ret != 0) { + flb_upstream_conn_release(u_conn); + + flb_plg_error(ctx->ins, + "cannot compress payload, aborting"); + + return FLB_ERROR; + } + + /* Create HTTP client context */ + c = flb_http_client(u_conn, FLB_HTTP_POST, ctx->uri, + payload_buf, payload_size, + ctx->host, ctx->port, + ctx->proxy, 0); + + + if (c->proxy.host) { + flb_plg_debug(ctx->ins, "[http_client] proxy host: %s port: %i", + c->proxy.host, c->proxy.port); + } + + /* Allow duplicated headers ? */ + flb_http_allow_duplicated_headers(c, FLB_FALSE); + + /* + * Direct assignment of the callback context to the HTTP client context. + * This needs to be improved through a more clean API. + */ + c->cb_ctx = ctx->ins->callback; + + flb_http_add_header(c, + FLB_PROMETHEUS_REMOTE_WRITE_CONTENT_TYPE_HEADER_NAME, + sizeof(FLB_PROMETHEUS_REMOTE_WRITE_CONTENT_TYPE_HEADER_NAME) - 1, + FLB_PROMETHEUS_REMOTE_WRITE_MIME_PROTOBUF_LITERAL, + sizeof(FLB_PROMETHEUS_REMOTE_WRITE_MIME_PROTOBUF_LITERAL) - 1); + + flb_http_add_header(c, + FLB_PROMETHEUS_REMOTE_WRITE_VERSION_HEADER_NAME, + sizeof(FLB_PROMETHEUS_REMOTE_WRITE_VERSION_HEADER_NAME) - 1, + FLB_PROMETHEUS_REMOTE_WRITE_VERSION_LITERAL, + sizeof(FLB_PROMETHEUS_REMOTE_WRITE_VERSION_LITERAL) - 1); + + if (strcasecmp(ctx->compression, "snappy") == 0) { + flb_http_add_header(c, + "Content-Encoding", + strlen("Content-Encoding"), + "snappy", + strlen("snappy")); + } + else if (strcasecmp(ctx->compression, "gzip") == 0) { + flb_http_add_header(c, + "Content-Encoding", + strlen("Content-Encoding"), + "gzip", + strlen("gzip")); + } + + /* Basic Auth headers */ + if (ctx->http_user && ctx->http_passwd) { + flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd); + } + + flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); + + flb_config_map_foreach(head, mv, ctx->headers) { + key = mk_list_entry_first(mv->val.list, struct flb_slist_entry, _head); + val = mk_list_entry_last(mv->val.list, struct flb_slist_entry, _head); + + flb_http_add_header(c, + key->str, flb_sds_len(key->str), + val->str, flb_sds_len(val->str)); + } + +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS + /* AWS SigV4 headers */ + if (ctx->has_aws_auth == FLB_TRUE) { + flb_plg_debug(ctx->ins, "signing request with AWS Sigv4"); + signature = flb_signv4_do(c, + FLB_TRUE, /* normalize URI ? */ + FLB_TRUE, /* add x-amz-date header ? */ + time(NULL), + (char *) ctx->aws_region, + (char *) ctx->aws_service, + 0, NULL, + ctx->aws_provider); + + if (!signature) { + flb_plg_error(ctx->ins, "could not sign request with sigv4"); + out_ret = FLB_RETRY; + goto cleanup; + } + flb_sds_destroy(signature); + } +#endif +#endif + + ret = flb_http_do(c, &b_sent); + if (ret == 0) { + /* + * Only allow the following HTTP status: + * + * - 200: OK + * - 201: Created + * - 202: Accepted + * - 203: no authorative resp + * - 204: No Content + * - 205: Reset content + * + */ + if ((c->resp.status < 200 || c->resp.status > 205) && + c->resp.status != 400) { + if (ctx->log_response_payload && + c->resp.payload && c->resp.payload_size > 0) { + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i\n%s", + ctx->host, ctx->port, + c->resp.status, c->resp.payload); + } + else { + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i", + ctx->host, ctx->port, c->resp.status); + } + out_ret = FLB_RETRY; + } + else if (c->resp.status == 400) { + /* Returned 400 status means unrecoverable. Immidiately + * returning as a error. */ + if (ctx->log_response_payload && + c->resp.payload && c->resp.payload_size > 0) { + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i\n%s", + ctx->host, ctx->port, + c->resp.status, c->resp.payload); + } + else { + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i", + ctx->host, ctx->port, c->resp.status); + } + out_ret = FLB_ERROR; + } + else { + if (ctx->log_response_payload && + c->resp.payload && c->resp.payload_size > 0) { + flb_plg_debug(ctx->ins, "%s:%i, HTTP status=%i\n%s", + ctx->host, ctx->port, + c->resp.status, c->resp.payload); + } + else { + flb_plg_debug(ctx->ins, "%s:%i, HTTP status=%i", + ctx->host, ctx->port, + c->resp.status); + } + } + } + else { + flb_plg_error(ctx->ins, "could not flush records to %s:%i (http_do=%i)", + ctx->host, ctx->port, ret); + out_ret = FLB_RETRY; + } + +cleanup: + /* + * If the payload buffer is different than incoming records in body, means + * we generated a different payload and must be freed. + */ + if (payload_buf != body) { + flb_free(payload_buf); + } + + /* Destroy HTTP client context */ + flb_http_client_destroy(c); + + /* Release the TCP connection */ + flb_upstream_conn_release(u_conn); + + return out_ret; +} + +static int cb_prom_init(struct flb_output_instance *ins, + struct flb_config *config, + void *data) +{ + struct prometheus_remote_write_context *ctx; + + ctx = flb_prometheus_remote_write_context_create(ins, config); + if (!ctx) { + return -1; + } + + flb_output_set_context(ins, ctx); + + return 0; +} + +static void append_labels(struct prometheus_remote_write_context *ctx, + struct cmt *cmt) +{ + struct flb_kv *kv; + struct mk_list *head; + + mk_list_foreach(head, &ctx->kv_labels) { + kv = mk_list_entry(head, struct flb_kv, _head); + cmt_label_add(cmt, kv->key, kv->val); + } +} + +static void cb_prom_flush(struct flb_event_chunk *event_chunk, + struct flb_output_flush *out_flush, + struct flb_input_instance *ins, void *out_context, + struct flb_config *config) +{ + int c = 0; + int ok; + int ret; + int result; + cfl_sds_t encoded_chunk; + flb_sds_t buf = NULL; + size_t diff = 0; + size_t off = 0; + struct cmt *cmt; + struct prometheus_remote_write_context *ctx = out_context; + + /* Initialize vars */ + ctx = out_context; + ok = CMT_DECODE_MSGPACK_SUCCESS; + result = FLB_OK; + + /* Buffer to concatenate multiple metrics contexts */ + buf = flb_sds_create_size(event_chunk->size); + if (!buf) { + flb_plg_error(ctx->ins, "could not allocate outgoing buffer"); + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + flb_plg_debug(ctx->ins, "cmetrics msgpack size: %lu", + event_chunk->size); + + /* Decode and encode every CMetric context */ + diff = 0; + while ((ret = cmt_decode_msgpack_create(&cmt, + (char *) event_chunk->data, + event_chunk->size, &off)) == ok) { + /* append labels set by config */ + append_labels(ctx, cmt); + + /* Create a Prometheus Remote Write payload */ + encoded_chunk = cmt_encode_prometheus_remote_write_create(cmt); + if (encoded_chunk == NULL) { + flb_plg_error(ctx->ins, + "Error encoding context as prometheus remote write"); + result = FLB_ERROR; + goto exit; + } + + flb_plg_debug(ctx->ins, "cmetric_id=%i decoded %lu-%lu payload_size=%lu", + c, diff, off, flb_sds_len(encoded_chunk)); + c++; + diff = off; + + /* concat buffer */ + flb_sds_cat_safe(&buf, encoded_chunk, flb_sds_len(encoded_chunk)); + + /* release */ + cmt_encode_prometheus_remote_write_destroy(encoded_chunk); + cmt_destroy(cmt); + } + + if (ret == CMT_DECODE_MSGPACK_INSUFFICIENT_DATA && c > 0) { + flb_plg_debug(ctx->ins, "final payload size: %lu", flb_sds_len(buf)); + if (buf && flb_sds_len(buf) > 0) { + /* Send HTTP request */ + result = http_post(ctx, buf, flb_sds_len(buf), + event_chunk->tag, + flb_sds_len(event_chunk->tag)); + + /* Debug http_post() result statuses */ + if (result == FLB_OK) { + flb_plg_debug(ctx->ins, "http_post result FLB_OK"); + } + else if (result == FLB_ERROR) { + flb_plg_debug(ctx->ins, "http_post result FLB_ERROR"); + } + else if (result == FLB_RETRY) { + flb_plg_debug(ctx->ins, "http_post result FLB_RETRY"); + } + } + flb_sds_destroy(buf); + buf = NULL; + } + else { + flb_plg_error(ctx->ins, "Error decoding msgpack encoded context"); + } + +exit: + if (buf) { + flb_sds_destroy(buf); + } + FLB_OUTPUT_RETURN(result); +} + +static int cb_prom_exit(void *data, struct flb_config *config) +{ + struct prometheus_remote_write_context *ctx; + + ctx = (struct prometheus_remote_write_context *) data; + + flb_prometheus_remote_write_context_destroy(ctx); + + return 0; +} + +/* Configuration properties map */ +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_SLIST_1, "add_label", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct prometheus_remote_write_context, + add_labels), + "Adds a custom label to the metrics use format: 'add_label name value'" + }, + + { + FLB_CONFIG_MAP_STR, "proxy", NULL, + 0, FLB_FALSE, 0, + "Specify an HTTP Proxy. The expected format of this value is http://host:port. " + }, + { + FLB_CONFIG_MAP_STR, "http_user", NULL, + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, http_user), + "Set HTTP auth user" + }, + { + FLB_CONFIG_MAP_STR, "http_passwd", "", + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, http_passwd), + "Set HTTP auth password" + }, + { + FLB_CONFIG_MAP_STR, "compression", "snappy", + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, compression), + "Compress the payload with either snappy, gzip if set" + }, + +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS + { + FLB_CONFIG_MAP_BOOL, "aws_auth", "false", + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, has_aws_auth), + "Enable AWS SigV4 authentication" + }, + { + FLB_CONFIG_MAP_STR, "aws_service", "aps", + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, aws_service), + "AWS destination service code, used by SigV4 authentication" + }, + FLB_AWS_CREDENTIAL_BASE_CONFIG_MAP(FLB_PROMETHEUS_REMOTE_WRITE_CREDENTIAL_PREFIX), +#endif +#endif + { + FLB_CONFIG_MAP_SLIST_1, "header", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct prometheus_remote_write_context, headers), + "Add a HTTP header key/value pair. Multiple headers can be set" + }, + { + FLB_CONFIG_MAP_STR, "uri", NULL, + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, uri), + "Specify an optional HTTP URI for the target web server, e.g: /something" + }, + { + FLB_CONFIG_MAP_BOOL, "log_response_payload", "true", + 0, FLB_TRUE, offsetof(struct prometheus_remote_write_context, log_response_payload), + "Specify if the response paylod should be logged or not" + }, + /* EOF */ + {0} +}; + +/* Plugin reference */ +struct flb_output_plugin out_prometheus_remote_write_plugin = { + .name = "prometheus_remote_write", + .description = "Prometheus remote write", + .cb_init = cb_prom_init, + .cb_flush = cb_prom_flush, + .cb_exit = cb_prom_exit, + .config_map = config_map, + .event_type = FLB_OUTPUT_METRICS, + .workers = 2, + .flags = FLB_OUTPUT_NET | FLB_IO_OPT_TLS, +}; diff --git a/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write.h b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write.h new file mode 100644 index 000000000..349c7a7b0 --- /dev/null +++ b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write.h @@ -0,0 +1,83 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FLB_PROMETHEUS_REMOTE_WRITE_H +#define FLB_PROMETHEUS_REMOTE_WRITE_H + +#include <fluent-bit/flb_output_plugin.h> + +#define FLB_PROMETHEUS_REMOTE_WRITE_CONTENT_TYPE_HEADER_NAME "Content-Type" +#define FLB_PROMETHEUS_REMOTE_WRITE_MIME_PROTOBUF_LITERAL "application/x-protobuf" +#define FLB_PROMETHEUS_REMOTE_WRITE_VERSION_HEADER_NAME "X-Prometheus-Remote-Write-Version" +#define FLB_PROMETHEUS_REMOTE_WRITE_VERSION_LITERAL "0.1.0" +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS +#define FLB_PROMETHEUS_REMOTE_WRITE_CREDENTIAL_PREFIX "aws_" +#endif +#endif + +/* Plugin context */ +struct prometheus_remote_write_context { + /* HTTP Auth */ + char *http_user; + char *http_passwd; + + /* AWS Auth */ +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS + int has_aws_auth; + struct flb_aws_provider *aws_provider; + const char *aws_region; + const char *aws_service; +#endif +#endif + + /* Proxy */ + const char *proxy; + char *proxy_host; + int proxy_port; + + /* HTTP URI */ + char *uri; + char *host; + int port; + + const char *compression; + + /* Log the response paylod */ + int log_response_payload; + + /* config reader for 'add_label' */ + struct mk_list *add_labels; + + /* internal labels ready to append */ + struct mk_list kv_labels; + + /* Upstream connection to the backend server */ + struct flb_upstream *u; + + /* Arbitrary HTTP headers */ + struct mk_list *headers; + + + /* instance context */ + struct flb_output_instance *ins; +}; + +#endif diff --git a/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write_conf.c b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write_conf.c new file mode 100644 index 000000000..09c7e0d52 --- /dev/null +++ b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write_conf.c @@ -0,0 +1,254 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <fluent-bit/flb_output_plugin.h> +#include <fluent-bit/flb_utils.h> +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_sds.h> +#include <fluent-bit/flb_kv.h> +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS +#include <fluent-bit/flb_aws_credentials.h> +#endif +#endif +#include "remote_write.h" +#include "remote_write_conf.h" + +static int config_add_labels(struct flb_output_instance *ins, + struct prometheus_remote_write_context *ctx) +{ + struct mk_list *head; + struct flb_config_map_val *mv; + struct flb_slist_entry *k = NULL; + struct flb_slist_entry *v = NULL; + struct flb_kv *kv; + + if (!ctx->add_labels || mk_list_size(ctx->add_labels) == 0) { + return 0; + } + + /* iterate all 'add_label' definitions */ + flb_config_map_foreach(head, mv, ctx->add_labels) { + if (mk_list_size(mv->val.list) != 2) { + flb_plg_error(ins, "'add_label' expects a key and a value, " + "e.g: 'add_label version 1.8.0'"); + return -1; + } + + k = mk_list_entry_first(mv->val.list, struct flb_slist_entry, _head); + v = mk_list_entry_last(mv->val.list, struct flb_slist_entry, _head); + + kv = flb_kv_item_create(&ctx->kv_labels, k->str, v->str); + if (!kv) { + flb_plg_error(ins, "could not append label %s=%s\n", k->str, v->str); + return -1; + } + } + + return 0; +} + +struct prometheus_remote_write_context *flb_prometheus_remote_write_context_create( + struct flb_output_instance *ins, struct flb_config *config) +{ + int ret; + int ulen; + int io_flags = 0; + char *protocol = NULL; + char *host = NULL; + char *port = NULL; + char *uri = NULL; + char *tmp_uri = NULL; + const char *tmp; + struct flb_upstream *upstream; + struct prometheus_remote_write_context *ctx = NULL; + + /* Allocate plugin context */ + ctx = flb_calloc(1, sizeof(struct prometheus_remote_write_context)); + if (!ctx) { + flb_errno(); + return NULL; + } + ctx->ins = ins; + mk_list_init(&ctx->kv_labels); + + ret = flb_output_config_map_set(ins, (void *) ctx); + if (ret == -1) { + flb_free(ctx); + return NULL; + } + + /* Parse 'add_label' */ + ret = config_add_labels(ins, ctx); + if (ret == -1) { + return NULL; + } + + /* + * Check if a Proxy have been set, if so the Upstream manager will use + * the Proxy end-point and then we let the HTTP client know about it, so + * it can adjust the HTTP requests. + */ + tmp = flb_output_get_property("proxy", ins); + if (tmp) { + ret = flb_utils_url_split(tmp, &protocol, &host, &port, &uri); + if (ret == -1) { + flb_plg_error(ctx->ins, "could not parse proxy parameter: '%s'", tmp); + flb_free(ctx); + return NULL; + } + + ctx->proxy_host = host; + ctx->proxy_port = atoi(port); + ctx->proxy = tmp; + flb_free(protocol); + flb_free(port); + flb_free(uri); + uri = NULL; + } + else { + flb_output_net_default("127.0.0.1", 80, ins); + } + + /* Check if AWS SigV4 authentication is enabled */ +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS + if (ctx->has_aws_auth) { + if (!ctx->aws_service) { + flb_plg_error(ins, "aws_auth option requires " FLB_PROMETHEUS_REMOTE_WRITE_CREDENTIAL_PREFIX + "service to be set"); + flb_free(ctx); + return NULL; + } + + ctx->aws_provider = flb_managed_chain_provider_create( + ins, + config, + FLB_PROMETHEUS_REMOTE_WRITE_CREDENTIAL_PREFIX, + NULL, + flb_aws_client_generator() + ); + if (!ctx->aws_provider) { + flb_plg_error(ins, "failed to create aws credential provider for sigv4 auth"); + flb_free(ctx); + return NULL; + } + + /* If managed provider creation succeeds, then region key is present */ + ctx->aws_region = flb_output_get_property(FLB_PROMETHEUS_REMOTE_WRITE_CREDENTIAL_PREFIX + "region", ctx->ins); + } +#endif /* !FLB_HAVE_AWS */ +#endif /* !FLB_HAVE_SIGNV4 */ + + /* Check if SSL/TLS is enabled */ +#ifdef FLB_HAVE_TLS + if (ins->use_tls == FLB_TRUE) { + io_flags = FLB_IO_TLS; + } + else { + io_flags = FLB_IO_TCP; + } +#else + io_flags = FLB_IO_TCP; +#endif + + if (ins->host.ipv6 == FLB_TRUE) { + io_flags |= FLB_IO_IPV6; + } + + if (ctx->proxy) { + flb_plg_trace(ctx->ins, "Upstream Proxy=%s:%i", + ctx->proxy_host, ctx->proxy_port); + upstream = flb_upstream_create(config, + ctx->proxy_host, + ctx->proxy_port, + io_flags, ins->tls); + } + else { + upstream = flb_upstream_create(config, + ins->host.name, + ins->host.port, + io_flags, ins->tls); + } + + if (!upstream) { + flb_free(ctx); + return NULL; + } + + if (ins->host.uri) { + uri = flb_strdup(ins->host.uri->full); + } + else { + tmp = flb_output_get_property("uri", ins); + if (tmp) { + uri = flb_strdup(tmp); + } + } + + if (!uri) { + uri = flb_strdup("/"); + } + else if (uri[0] != '/') { + ulen = strlen(uri); + tmp_uri = flb_malloc(ulen + 2); + tmp_uri[0] = '/'; + memcpy(tmp_uri + 1, uri, ulen); + tmp_uri[ulen + 1] = '\0'; + flb_free(uri); + uri = tmp_uri; + } + + ctx->u = upstream; + ctx->uri = uri; + ctx->host = ins->host.name; + ctx->port = ins->host.port; + + /* Set instance flags into upstream */ + flb_output_upstream_set(ctx->u, ins); + + return ctx; +} + +void flb_prometheus_remote_write_context_destroy( + struct prometheus_remote_write_context *ctx) +{ + if (!ctx) { + return; + } + + flb_kv_release(&ctx->kv_labels); + + if (ctx->u) { + flb_upstream_destroy(ctx->u); + } + +#ifdef FLB_HAVE_SIGNV4 +#ifdef FLB_HAVE_AWS + if (ctx->aws_provider) { + flb_aws_provider_destroy(ctx->aws_provider); + } +#endif +#endif + + flb_free(ctx->proxy_host); + flb_free(ctx->uri); + flb_free(ctx); +} diff --git a/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write_conf.h b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write_conf.h new file mode 100644 index 000000000..e6268fc54 --- /dev/null +++ b/src/fluent-bit/plugins/out_prometheus_remote_write/remote_write_conf.h @@ -0,0 +1,33 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FLB_OUT_PROMETHEUS_REMOTE_WRITE_CONF_H +#define FLB_OUT_PROMETHEUS_REMOTE_WRITE_CONF_H + +#include <fluent-bit/flb_info.h> +#include <fluent-bit/flb_output.h> + +#include "remote_write.h" + +struct prometheus_remote_write_context *flb_prometheus_remote_write_context_create( + struct flb_output_instance *ins, struct flb_config *config); +void flb_prometheus_remote_write_context_destroy( + struct prometheus_remote_write_context *ctx); + +#endif |