diff options
Diffstat (limited to 'src/fluent-bit/plugins/out_azure_kusto')
7 files changed, 1814 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/out_azure_kusto/CMakeLists.txt b/src/fluent-bit/plugins/out_azure_kusto/CMakeLists.txt new file mode 100644 index 000000000..6803bee09 --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/CMakeLists.txt @@ -0,0 +1,7 @@ +set(src + azure_kusto.c + azure_kusto_conf.c + azure_kusto_ingest.c + ) + +FLB_PLUGIN(out_azure_kusto "${src}" "") diff --git a/src/fluent-bit/plugins/out_azure_kusto/azure_kusto.c b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto.c new file mode 100644 index 000000000..4b8ad9b82 --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto.c @@ -0,0 +1,477 @@ +/* -*- 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_http_client.h> +#include <fluent-bit/flb_kv.h> +#include <fluent-bit/flb_oauth2.h> +#include <fluent-bit/flb_output_plugin.h> +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_signv4.h> +#include <fluent-bit/flb_log_event_decoder.h> + +#include "azure_kusto.h" +#include "azure_kusto_conf.h" +#include "azure_kusto_ingest.h" + +/* Create a new oauth2 context and get a oauth2 token */ +static int azure_kusto_get_oauth2_token(struct flb_azure_kusto *ctx) +{ + int ret; + char *token; + + /* Clear any previous oauth2 payload content */ + flb_oauth2_payload_clear(ctx->o); + + ret = flb_oauth2_payload_append(ctx->o, "grant_type", 10, "client_credentials", 18); + if (ret == -1) { + flb_plg_error(ctx->ins, "error appending oauth2 params"); + return -1; + } + + ret = flb_oauth2_payload_append(ctx->o, "scope", 5, FLB_AZURE_KUSTO_SCOPE, 39); + if (ret == -1) { + flb_plg_error(ctx->ins, "error appending oauth2 params"); + return -1; + } + + ret = flb_oauth2_payload_append(ctx->o, "client_id", 9, ctx->client_id, -1); + if (ret == -1) { + flb_plg_error(ctx->ins, "error appending oauth2 params"); + return -1; + } + + ret = flb_oauth2_payload_append(ctx->o, "client_secret", 13, ctx->client_secret, -1); + if (ret == -1) { + flb_plg_error(ctx->ins, "error appending oauth2 params"); + return -1; + } + + /* Retrieve access token */ + token = flb_oauth2_token_get(ctx->o); + if (!token) { + flb_plg_error(ctx->ins, "error retrieving oauth2 access token"); + return -1; + } + + return 0; +} + +flb_sds_t get_azure_kusto_token(struct flb_azure_kusto *ctx) +{ + int ret = 0; + flb_sds_t output = NULL; + + if (pthread_mutex_lock(&ctx->token_mutex)) { + flb_plg_error(ctx->ins, "error locking mutex"); + return NULL; + } + + if (flb_oauth2_token_expired(ctx->o) == FLB_TRUE) { + ret = azure_kusto_get_oauth2_token(ctx); + } + + /* Copy string to prevent race conditions (get_oauth2 can free the string) */ + if (ret == 0) { + output = flb_sds_create_size(flb_sds_len(ctx->o->token_type) + + flb_sds_len(ctx->o->access_token) + 2); + if (!output) { + flb_plg_error(ctx->ins, "error creating token buffer"); + return NULL; + } + flb_sds_snprintf(&output, flb_sds_alloc(output), "%s %s", ctx->o->token_type, + ctx->o->access_token); + } + + if (pthread_mutex_unlock(&ctx->token_mutex)) { + flb_plg_error(ctx->ins, "error unlocking mutex"); + if (output) { + flb_sds_destroy(output); + } + return NULL; + } + + return output; +} + +/** + * Executes a control command against kusto's endpoint + * + * @param ctx Plugin's context + * @param csl Kusto's control command + * @return flb_sds_t Returns the response or NULL on error. + */ +flb_sds_t execute_ingest_csl_command(struct flb_azure_kusto *ctx, const char *csl) +{ + flb_sds_t token; + flb_sds_t body; + size_t b_sent; + int ret; + struct flb_connection *u_conn; + struct flb_http_client *c; + flb_sds_t resp = NULL; + + /* Get upstream connection */ + u_conn = flb_upstream_conn_get(ctx->u); + + if (u_conn) { + token = get_azure_kusto_token(ctx); + + if (token) { + /* Compose request body */ + body = flb_sds_create_size(sizeof(FLB_AZURE_KUSTO_MGMT_BODY_TEMPLATE) - 1 + + strlen(csl)); + + if (body) { + flb_sds_snprintf(&body, flb_sds_alloc(body), + FLB_AZURE_KUSTO_MGMT_BODY_TEMPLATE, csl); + + /* Compose HTTP Client request */ + c = flb_http_client(u_conn, FLB_HTTP_POST, FLB_AZURE_KUSTO_MGMT_URI_PATH, + body, flb_sds_len(body), NULL, 0, NULL, 0); + + if (c) { + /* Add headers */ + flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); + flb_http_add_header(c, "Content-Type", 12, "application/json", 16); + flb_http_add_header(c, "Accept", 6, "application/json", 16); + flb_http_add_header(c, "Authorization", 13, token, + flb_sds_len(token)); + flb_http_buffer_size(c, FLB_HTTP_DATA_SIZE_MAX * 10); + + /* Send HTTP request */ + ret = flb_http_do(c, &b_sent); + flb_plg_debug( + ctx->ins, + "Kusto ingestion command request http_do=%i, HTTP Status: %i", + ret, c->resp.status); + + if (ret == 0) { + if (c->resp.status == 200) { + /* Copy payload response to the response param */ + resp = + flb_sds_create_len(c->resp.payload, c->resp.payload_size); + } + else if (c->resp.payload_size > 0) { + flb_plg_debug(ctx->ins, "Request failed and returned: \n%s", + c->resp.payload); + } + else { + flb_plg_debug(ctx->ins, "Request failed"); + } + } + else { + flb_plg_error(ctx->ins, "cannot send HTTP request"); + } + + flb_http_client_destroy(c); + } + else { + flb_plg_error(ctx->ins, "cannot create HTTP client context"); + } + + flb_sds_destroy(body); + } + else { + flb_plg_error(ctx->ins, "cannot construct request body"); + } + + flb_sds_destroy(token); + } + else { + flb_plg_error(ctx->ins, "cannot retrieve oauth2 token"); + } + + flb_upstream_conn_release(u_conn); + } + else { + flb_plg_error(ctx->ins, "cannot create upstream connection"); + } + + return resp; +} + +static int cb_azure_kusto_init(struct flb_output_instance *ins, struct flb_config *config, + void *data) +{ + int io_flags = FLB_IO_TLS; + struct flb_azure_kusto *ctx; + + /* Create config context */ + ctx = flb_azure_kusto_conf_create(ins, config); + if (!ctx) { + flb_plg_error(ins, "configuration failed"); + return -1; + } + + flb_output_set_context(ins, ctx); + + /* Network mode IPv6 */ + if (ins->host.ipv6 == FLB_TRUE) { + io_flags |= FLB_IO_IPV6; + } + + /* Create mutex for acquiring oauth tokens and getting ingestion resources (they + * are shared across flush coroutines) + */ + pthread_mutex_init(&ctx->token_mutex, NULL); + pthread_mutex_init(&ctx->resources_mutex, NULL); + + /* + * Create upstream context for Kusto Ingestion endpoint + */ + ctx->u = flb_upstream_create_url(config, ctx->ingestion_endpoint, io_flags, ins->tls); + if (!ctx->u) { + flb_plg_error(ctx->ins, "upstream creation failed"); + return -1; + } + + /* Create oauth2 context */ + ctx->o = + flb_oauth2_create(ctx->config, ctx->oauth_url, FLB_AZURE_KUSTO_TOKEN_REFRESH); + if (!ctx->o) { + flb_plg_error(ctx->ins, "cannot create oauth2 context"); + return -1; + } + flb_output_upstream_set(ctx->u, ins); + + return 0; +} + +static int azure_kusto_format(struct flb_azure_kusto *ctx, const char *tag, int tag_len, + const void *data, size_t bytes, void **out_data, + size_t *out_size) +{ + int records = 0; + msgpack_sbuffer mp_sbuf; + msgpack_packer mp_pck; + /* for sub msgpack objs */ + int map_size; + struct tm tms; + char time_formatted[32]; + size_t s; + int len; + struct flb_log_event_decoder log_decoder; + struct flb_log_event log_event; + int ret; + /* output buffer */ + flb_sds_t out_buf; + + /* Create array for all records */ + records = flb_mp_count(data, bytes); + if (records <= 0) { + flb_plg_error(ctx->ins, "error counting msgpack entries"); + return -1; + } + + ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes); + + if (ret != FLB_EVENT_DECODER_SUCCESS) { + flb_plg_error(ctx->ins, + "Log event decoder initialization error : %d", ret); + + return -1; + } + + /* Create temporary msgpack buffer */ + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + + msgpack_pack_array(&mp_pck, records); + + while ((ret = flb_log_event_decoder_next( + &log_decoder, + &log_event)) == FLB_EVENT_DECODER_SUCCESS) { + map_size = 1; + if (ctx->include_time_key == FLB_TRUE) { + map_size++; + } + + if (ctx->include_tag_key == FLB_TRUE) { + map_size++; + } + + msgpack_pack_map(&mp_pck, map_size); + + /* include_time_key */ + if (ctx->include_time_key == FLB_TRUE) { + msgpack_pack_str(&mp_pck, flb_sds_len(ctx->time_key)); + msgpack_pack_str_body(&mp_pck, ctx->time_key, flb_sds_len(ctx->time_key)); + + /* Append the time value as ISO 8601 */ + gmtime_r(&log_event.timestamp.tm.tv_sec, &tms); + s = strftime(time_formatted, sizeof(time_formatted) - 1, + FLB_PACK_JSON_DATE_ISO8601_FMT, &tms); + + len = snprintf(time_formatted + s, sizeof(time_formatted) - 1 - s, + ".%03" PRIu64 "Z", + (uint64_t)log_event.timestamp.tm.tv_nsec / 1000000); + s += len; + msgpack_pack_str(&mp_pck, s); + msgpack_pack_str_body(&mp_pck, time_formatted, s); + } + + /* include_tag_key */ + if (ctx->include_tag_key == FLB_TRUE) { + msgpack_pack_str(&mp_pck, flb_sds_len(ctx->tag_key)); + msgpack_pack_str_body(&mp_pck, ctx->tag_key, flb_sds_len(ctx->tag_key)); + msgpack_pack_str(&mp_pck, tag_len); + msgpack_pack_str_body(&mp_pck, tag, tag_len); + } + + msgpack_pack_str(&mp_pck, flb_sds_len(ctx->log_key)); + msgpack_pack_str_body(&mp_pck, ctx->log_key, flb_sds_len(ctx->log_key)); + msgpack_pack_object(&mp_pck, *log_event.body); + } + + /* Convert from msgpack to JSON */ + out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size); + + /* Cleanup */ + flb_log_event_decoder_destroy(&log_decoder); + msgpack_sbuffer_destroy(&mp_sbuf); + + if (!out_buf) { + flb_plg_error(ctx->ins, "error formatting JSON payload"); + return -1; + } + + *out_data = out_buf; + *out_size = flb_sds_len(out_buf); + + return 0; +} + +static void cb_azure_kusto_flush(struct flb_event_chunk *event_chunk, + struct flb_output_flush *out_flush, + struct flb_input_instance *i_ins, void *out_context, + struct flb_config *config) +{ + int ret; + flb_sds_t json; + size_t json_size; + size_t tag_len; + struct flb_azure_kusto *ctx = out_context; + + (void)i_ins; + (void)config; + + flb_plg_trace(ctx->ins, "flushing bytes %zu", event_chunk->size); + + tag_len = flb_sds_len(event_chunk->tag); + + /* Load or refresh ingestion resources */ + ret = azure_kusto_load_ingestion_resources(ctx, config); + if (ret != 0) { + flb_plg_error(ctx->ins, "cannot load ingestion resources"); + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + /* Reformat msgpack to JSON payload */ + ret = azure_kusto_format(ctx, event_chunk->tag, tag_len, event_chunk->data, + event_chunk->size, (void **)&json, &json_size); + if (ret != 0) { + flb_plg_error(ctx->ins, "cannot reformat data into json"); + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + ret = azure_kusto_queued_ingestion(ctx, event_chunk->tag, tag_len, json, json_size); + if (ret != 0) { + flb_plg_error(ctx->ins, "cannot perform queued ingestion"); + flb_sds_destroy(json); + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + /* Cleanup */ + flb_sds_destroy(json); + + /* Done */ + FLB_OUTPUT_RETURN(FLB_OK); +} + +static int cb_azure_kusto_exit(void *data, struct flb_config *config) +{ + struct flb_azure_kusto *ctx = data; + + if (!ctx) { + return -1; + } + + if (ctx->u) { + flb_upstream_destroy(ctx->u); + ctx->u = NULL; + } + + flb_azure_kusto_conf_destroy(ctx); + + return 0; +} + +static struct flb_config_map config_map[] = { + {FLB_CONFIG_MAP_STR, "tenant_id", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, tenant_id), + "Set the tenant ID of the AAD application used for authentication"}, + {FLB_CONFIG_MAP_STR, "client_id", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, client_id), + "Set the client ID (Application ID) of the AAD application used for authentication"}, + {FLB_CONFIG_MAP_STR, "client_secret", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, client_secret), + "Set the client secret (Application Password) of the AAD application used for " + "authentication"}, + {FLB_CONFIG_MAP_STR, "ingestion_endpoint", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, ingestion_endpoint), + "Set the Kusto cluster's ingestion endpoint URL (e.g. " + "https://ingest-mycluster.eastus.kusto.windows.net)"}, + {FLB_CONFIG_MAP_STR, "database_name", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, database_name), "Set the database name"}, + {FLB_CONFIG_MAP_STR, "table_name", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, table_name), "Set the table name"}, + {FLB_CONFIG_MAP_STR, "ingestion_mapping_reference", (char *)NULL, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, ingestion_mapping_reference), + "Set the ingestion mapping reference"}, + {FLB_CONFIG_MAP_STR, "log_key", FLB_AZURE_KUSTO_DEFAULT_LOG_KEY, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, log_key), "The key name of event payload"}, + {FLB_CONFIG_MAP_BOOL, "include_tag_key", "true", 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, include_tag_key), + "If enabled, tag is appended to output. " + "The key name is used 'tag_key' property."}, + {FLB_CONFIG_MAP_STR, "tag_key", FLB_AZURE_KUSTO_DEFAULT_TAG_KEY, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, tag_key), + "The key name of tag. If 'include_tag_key' is false, " + "This property is ignored"}, + {FLB_CONFIG_MAP_BOOL, "include_time_key", "true", 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, include_time_key), + "If enabled, time is appended to output. " + "The key name is used 'time_key' property."}, + {FLB_CONFIG_MAP_STR, "time_key", FLB_AZURE_KUSTO_DEFAULT_TIME_KEY, 0, FLB_TRUE, + offsetof(struct flb_azure_kusto, time_key), + "The key name of the time. If 'include_time_key' is false, " + "This property is ignored"}, + /* EOF */ + {0}}; + +struct flb_output_plugin out_azure_kusto_plugin = { + .name = "azure_kusto", + .description = "Send events to Kusto (Azure Data Explorer)", + .cb_init = cb_azure_kusto_init, + .cb_flush = cb_azure_kusto_flush, + .cb_exit = cb_azure_kusto_exit, + .config_map = config_map, + /* Plugin flags */ + .flags = FLB_OUTPUT_NET | FLB_IO_TLS, +}; diff --git a/src/fluent-bit/plugins/out_azure_kusto/azure_kusto.h b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto.h new file mode 100644 index 000000000..ac4eedfd0 --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto.h @@ -0,0 +1,110 @@ +/* -*- 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_AZURE_KUSTO +#define FLB_OUT_AZURE_KUSTO + +#include <fluent-bit/flb_info.h> +#include <fluent-bit/flb_oauth2.h> +#include <fluent-bit/flb_output.h> +#include <fluent-bit/flb_sds.h> +#include <fluent-bit/flb_upstream_ha.h> + +/* refresh token every 50 minutes */ +#define FLB_AZURE_KUSTO_TOKEN_REFRESH 3000 + +/* Kusto streaming inserts oauth scope */ +#define FLB_AZURE_KUSTO_SCOPE "https://help.kusto.windows.net/.default" + +/* MSAL authorization URL */ +#define FLB_MSAL_AUTH_URL_TEMPLATE \ + "https://login.microsoftonline.com/%s/oauth2/v2.0/token" + +#define FLB_AZURE_KUSTO_MGMT_URI_PATH "/v1/rest/mgmt" +#define FLB_AZURE_KUSTO_MGMT_BODY_TEMPLATE "{\"csl\":\"%s\", \"db\": \"NetDefaultDB\"}" + +#define FLB_AZURE_KUSTO_DEFAULT_TIME_KEY "timestamp" +#define FLB_AZURE_KUSTO_DEFAULT_TAG_KEY "tag" +#define FLB_AZURE_KUSTO_DEFAULT_LOG_KEY "log" + +#define AZURE_KUSTO_RESOURCE_STORAGE 0 +#define AZURE_KUSTO_RESOURCE_QUEUE 1 + +#define AZURE_KUSTO_RESOURCE_UPSTREAM_URI "uri" +#define AZURE_KUSTO_RESOURCE_UPSTREAM_SAS "sas" + +#define FLB_AZURE_KUSTO_RESOURCES_LOAD_INTERVAL_SEC 3600 + +struct flb_azure_kusto_resources { + struct flb_upstream_ha *blob_ha; + struct flb_upstream_ha *queue_ha; + flb_sds_t identity_token; + + /* used to reload resouces after some time */ + time_t load_time; +}; + +struct flb_azure_kusto { + /* azure_kusto configuration */ + flb_sds_t tenant_id; + flb_sds_t client_id; + flb_sds_t client_secret; + flb_sds_t ingestion_endpoint; + flb_sds_t database_name; + flb_sds_t table_name; + flb_sds_t ingestion_mapping_reference; + + /* records configuration */ + flb_sds_t log_key; + int include_tag_key; + flb_sds_t tag_key; + int include_time_key; + flb_sds_t time_key; + + /* --- internal data --- */ + + flb_sds_t ingestion_mgmt_endpoint; + + /* oauth2 context */ + flb_sds_t oauth_url; + struct flb_oauth2 *o; + + /* mutex for acquiring oauth tokens */ + pthread_mutex_t token_mutex; + + /* ingestion resources */ + struct flb_azure_kusto_resources *resources; + + /* mutex for loading reosurces */ + pthread_mutex_t resources_mutex; + + /* Upstream connection to the backend server */ + struct flb_upstream *u; + + /* Fluent Bit context */ + struct flb_config *config; + + /* Plugin output instance reference */ + struct flb_output_instance *ins; +}; + +flb_sds_t get_azure_kusto_token(struct flb_azure_kusto *ctx); +flb_sds_t execute_ingest_csl_command(struct flb_azure_kusto *ctx, const char *csl); + +#endif diff --git a/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_conf.c b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_conf.c new file mode 100644 index 000000000..5303fef67 --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_conf.c @@ -0,0 +1,665 @@ +/* -*- 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_info.h> +#include <fluent-bit/flb_jsmn.h> +#include <fluent-bit/flb_oauth2.h> +#include <fluent-bit/flb_output.h> +#include <fluent-bit/flb_output_plugin.h> +#include <fluent-bit/flb_sds.h> +#include <fluent-bit/flb_time.h> +#include <fluent-bit/flb_unescape.h> +#include <fluent-bit/flb_upstream_ha.h> +#include <fluent-bit/flb_utils.h> + +#include "azure_kusto.h" +#include "azure_kusto_conf.h" + +static struct flb_upstream_node *flb_upstream_node_create_url(struct flb_azure_kusto *ctx, + struct flb_config *config, + const char *url) +{ + int ret; + char *prot = NULL; + char *host = NULL; + char *port = NULL; + char *uri = NULL; + flb_sds_t sds_host = NULL; + flb_sds_t sds_port = NULL; + char *tmp; + struct flb_hash_table *kv = NULL; + struct flb_upstream_node *node = NULL; + int uri_length; + int sas_length; + + /* Parse and split URL */ + ret = flb_utils_url_split(url, &prot, &host, &port, &uri); + if (ret == -1) { + flb_plg_error(ctx->ins, "invalid URL: %s", url); + return NULL; + } + + /* find sas token in query */ + tmp = strchr(uri, '?'); + + if (tmp) { + uri_length = tmp - uri; + sas_length = strnlen(tmp + 1, 256); + + /* kv that will hold base uri, and sas token */ + kv = flb_hash_table_create(FLB_HASH_TABLE_EVICT_NONE, 2, 2); + + if (kv) { + ret = flb_hash_table_add(kv, AZURE_KUSTO_RESOURCE_UPSTREAM_URI, 3, uri, uri_length); + + if (ret != -1) { + ret = flb_hash_table_add(kv, AZURE_KUSTO_RESOURCE_UPSTREAM_SAS, 3, tmp + 1, + sas_length); + + if (ret != -1) { + /* if any/all of these creations would fail the node creation will fail and cleanup */ + sds_host = flb_sds_create(host); + sds_port = flb_sds_create(port); + + node = flb_upstream_node_create( + NULL, sds_host, sds_port, FLB_TRUE, ctx->ins->tls->verify, + ctx->ins->tls->debug, ctx->ins->tls->vhost, NULL, NULL, NULL, + NULL, NULL, kv, config); + + if (!node) { + flb_plg_error(ctx->ins, "error creating resource upstream node"); + } + } + else { + flb_plg_error(ctx->ins, "error storing resource sas token"); + } + } + else { + flb_plg_error(ctx->ins, "error storing resource uri"); + } + + /* avoid destorying if function is successful */ + if (!node) { + flb_hash_table_destroy(kv); + } + } + else { + flb_plg_error(ctx->ins, "error creating upstream node hash table"); + } + } + else { + flb_plg_error(ctx->ins, "uri has no sas token query: %s", uri); + } + + flb_free(prot); + flb_free(host); + flb_free(port); + flb_free(uri); + + return node; +} + +static int flb_azure_kusto_resources_clear(struct flb_azure_kusto_resources *resources) +{ + if (!resources) { + return -1; + } + + if (resources->blob_ha) { + flb_upstream_ha_destroy(resources->blob_ha); + resources->blob_ha = NULL; + } + + if (resources->queue_ha) { + flb_upstream_ha_destroy(resources->queue_ha); + resources->queue_ha = NULL; + } + + if (resources->identity_token) { + flb_sds_destroy(resources->identity_token); + resources->identity_token = NULL; + } + + resources->load_time = 0; + + return 0; +} + +/** + * Parses ".get ingestion resources" response into HA upstreams of the queue & blob + * resources in the response. + * + * @param ctx Pointer to the plugin's context + * @param config Pointer to the config + * @param response sds string containing the response body + * @param blob_ha Pointer to an HA upstream for the blob resources, that would be + * allocated here. + * @param queue_ha Pointer to an HA upstream for the queue resources, that would be + * allocated here. + * @return int 0 on success, -1 on failure + */ +static int parse_storage_resources(struct flb_azure_kusto *ctx, struct flb_config *config, + flb_sds_t response, struct flb_upstream_ha *blob_ha, + struct flb_upstream_ha *queue_ha) +{ + jsmn_parser parser; + jsmntok_t *t; + jsmntok_t *tokens; + int tok_size = 100; + int ret = -1; + int i; + int blob_count = 0; + int queue_count = 0; + char *token_str; + int token_str_len; + int resource_type; + struct flb_upstream_node *node; + struct flb_upstream_ha *ha; + flb_sds_t resource_uri; + + /* Response is a json in the form of + * { + * "Tables": [ + * { + * "TableName": "Table_0", + * "Columns": [...], + * "Rows": [ + * [ + * ("TempStorage" | "SecuredReadyForAggregationQueue" | + * "SuccessfulIngestionsQueue" | "FailedIngestionsQueue" | "IngestionsStatusTable"), + * <URI with SAS> + * ], + * ... + * ] + * } + * ] + * } + */ + + resource_uri = flb_sds_create(NULL); + if (!resource_uri) { + flb_plg_error(ctx->ins, "error allocating resource uri buffer"); + return -1; + } + + jsmn_init(&parser); + tokens = flb_calloc(1, sizeof(jsmntok_t) * tok_size); + + if (tokens) { + ret = jsmn_parse(&parser, response, flb_sds_len(response), tokens, tok_size); + + if (ret > 0) { + /* skip all tokens until we reach "Rows" */ + for (i = 0; i < ret - 1; i++) { + t = &tokens[i]; + + if (t->type != JSMN_STRING) { + continue; + } + + token_str = response + t->start; + token_str_len = (t->end - t->start); + + /** + * if we found the Rows key, skipping this token and the next one (key and + * wrapping array value) + */ + if (token_str_len == 4 && strncmp(token_str, "Rows", 4) == 0) { + i += 2; + break; + } + } + + /* iterating rows, each row will have 3 tokens: the array holding the column + * values, the first value containing the resource type, and the second value + * containing the resource uri */ + for (; i < ret; i++) { + t = &tokens[i]; + + /** + * each token should be an array with 2 strings: + * First will be the resource type (TempStorage, + * SecuredReadyForAggregationQueue, etc...) Second will be the SAS URI + */ + if (t->type != JSMN_ARRAY) { + break; + } + + /* move to the next token, first item in the array - resource type */ + i++; + t = &tokens[i]; + if (t->type != JSMN_STRING) { + break; + } + + token_str = response + t->start; + token_str_len = (t->end - t->start); + + flb_plg_debug(ctx->ins, "found resource of type: %.*s ", + t->end - t->start, response + t->start); + + if (token_str_len == 11 && strncmp(token_str, "TempStorage", 11) == 0) { + resource_type = AZURE_KUSTO_RESOURCE_STORAGE; + } + else if (token_str_len == 31 && + strncmp(token_str, "SecuredReadyForAggregationQueue", 31) == 0) { + resource_type = AZURE_KUSTO_RESOURCE_QUEUE; + } + /* we don't care about other resources so we just skip the next token and + move on to the next pair */ + else { + i++; + continue; + } + + /* move to the next token, second item in the array - resource URI */ + i++; + t = &tokens[i]; + + if (t->type != JSMN_STRING) { + break; + } + + token_str = response + t->start; + token_str_len = (t->end - t->start); + + resource_uri = flb_sds_copy(resource_uri, token_str, token_str_len); + if (resource_type == AZURE_KUSTO_RESOURCE_QUEUE) { + ha = queue_ha; + queue_count++; + } + else { + ha = blob_ha; + blob_count++; + } + + if (!ha) { + flb_plg_error(ctx->ins, "error creating HA upstream"); + ret = -1; + break; + } + + node = flb_upstream_node_create_url(ctx, config, resource_uri); + + if (!node) { + flb_plg_error(ctx->ins, "error creating HA upstream node"); + ret = -1; + break; + } + + flb_upstream_ha_node_add(ha, node); + } + + if (ret != -1) { + if (queue_count > 0 && blob_count > 0) { + flb_plg_debug(ctx->ins, + "parsed %d blob resources and %d queue resources", + blob_count, queue_count); + ret = 0; + } + else { + flb_plg_error(ctx->ins, "error parsing resources: missing resources"); + ret = -1; + } + } + } + else { + flb_plg_error(ctx->ins, "error parsing JSON response: %s", response); + ret = -1; + } + } + else { + flb_plg_error(ctx->ins, "error allocating tokens"); + ret = -1; + } + + flb_sds_destroy(resource_uri); + flb_free(tokens); + + return ret; +} + +/** + * Parses ".get kusto identity token" response and returns the token as an sds string + * + * @param ctx Pointer to the plugin's context + * @param response sds string containing the response body + * @return flb_sds_t The parsed token + */ +static flb_sds_t parse_ingestion_identity_token(struct flb_azure_kusto *ctx, + flb_sds_t response) +{ + flb_sds_t identity_token = NULL; + int tok_size = 19; + jsmn_parser parser; + jsmntok_t *t; + jsmntok_t *tokens; + int ret; + char *token_str; + int token_str_len; + + /** + * Response is a json in the form of + * { + * "Tables": [ + * { + * "TableName": "Table_0", + * "Columns": [{ + * "ColumnName": "AuthorizationContext", + * "DataType": "String", + * "ColumnType": "string" + * }], + * "Rows": [ + * [ + * <value>, + * ] + * ] + * } + * ] + * } + * i.e. only one row and one column is expected (exactly 13 tokens) and the value + * should be the last + */ + + jsmn_init(&parser); + tokens = flb_calloc(1, sizeof(jsmntok_t) * tok_size); + if (!tokens) { + flb_plg_error(ctx->ins, "error allocating tokens"); + return NULL; + } + + ret = jsmn_parse(&parser, response, flb_sds_len(response), tokens, tok_size); + if (ret > 0) { + t = &tokens[tok_size - 1]; + + if (t->type == JSMN_STRING) { + t = &tokens[tok_size - 1]; + token_str = response + t->start; + token_str_len = (t->end - t->start); + + identity_token = flb_sds_create_len(token_str, token_str_len); + + if (identity_token) { + flb_plg_debug(ctx->ins, "parsed kusto identity token: '%s'", + identity_token); + } + else { + flb_plg_error(ctx->ins, "error parsing kusto identity token"); + } + } + else { + flb_plg_error(ctx->ins, "unexpected JSON response: %s", response); + } + } + else { + flb_plg_error(ctx->ins, "error parsing JSON response: %s", response); + } + + flb_free(tokens); + + return identity_token; +} + +int azure_kusto_load_ingestion_resources(struct flb_azure_kusto *ctx, + struct flb_config *config) +{ + int ret = -1; + flb_sds_t response = NULL; + flb_sds_t identity_token = NULL; + struct flb_upstream_ha *blob_ha = NULL; + struct flb_upstream_ha *queue_ha = NULL; + time_t now; + + if (pthread_mutex_lock(&ctx->resources_mutex)) { + flb_plg_error(ctx->ins, "error locking mutex"); + return -1; + } + + now = time(NULL); + + /* check if we have all resources and they are not stale */ + if (ctx->resources->blob_ha && ctx->resources->queue_ha && + ctx->resources->identity_token && + now - ctx->resources->load_time < FLB_AZURE_KUSTO_RESOURCES_LOAD_INTERVAL_SEC) { + flb_plg_debug(ctx->ins, "resources are already loaded and are not stale"); + ret = 0; + } + else { + flb_plg_info(ctx->ins, "loading kusto ingestion resourcs"); + response = execute_ingest_csl_command(ctx, ".get ingestion resources"); + + if (response) { + queue_ha = flb_upstream_ha_create("azure_kusto_queue_ha"); + + if (queue_ha) { + blob_ha = flb_upstream_ha_create("azure_kusto_blob_ha"); + + if (blob_ha) { + ret = + parse_storage_resources(ctx, config, response, blob_ha, queue_ha); + + if (ret == 0) { + flb_sds_destroy(response); + response = NULL; + + response = + execute_ingest_csl_command(ctx, ".get kusto identity token"); + + if (response) { + identity_token = + parse_ingestion_identity_token(ctx, response); + + if (identity_token) { + ret = flb_azure_kusto_resources_clear(ctx->resources); + + if (ret != -1) { + ctx->resources->blob_ha = blob_ha; + ctx->resources->queue_ha = queue_ha; + ctx->resources->identity_token = identity_token; + ctx->resources->load_time = now; + + ret = 0; + } + else { + flb_plg_error( + ctx->ins, + "error destroying previous ingestion resources"); + } + } + else { + flb_plg_error(ctx->ins, + "error parsing ingestion identity token"); + ret = -1; + } + } + else { + flb_plg_error(ctx->ins, "error getting kusto identity token"); + ret = -1; + } + } + else { + flb_plg_error(ctx->ins, + "error parsing ingestion storage resources"); + ret = -1; + } + + if (ret == -1) { + flb_upstream_ha_destroy(blob_ha); + } + } + else { + flb_plg_error(ctx->ins, "error creating storage resources upstreams"); + ret = -1; + } + + if (ret == -1) { + flb_upstream_ha_destroy(queue_ha); + } + } + else { + flb_plg_error(ctx->ins, "error creating storage resources upstreams"); + } + + if (response) { + flb_sds_destroy(response); + } + } + if (!response) { + flb_plg_error(ctx->ins, "error getting ingestion storage resources"); + } + } + + if (pthread_mutex_unlock(&ctx->resources_mutex)) { + flb_plg_error(ctx->ins, "error unlocking mutex"); + return -1; + } + + return ret; +} + +static int flb_azure_kusto_resources_destroy(struct flb_azure_kusto_resources *resources) +{ + int ret; + + if (!resources) { + return -1; + } + + ret = flb_azure_kusto_resources_clear(resources); + if (ret != 0) { + return -1; + } + + flb_free(resources); + + return 0; +} + +struct flb_azure_kusto *flb_azure_kusto_conf_create(struct flb_output_instance *ins, + struct flb_config *config) +{ + int ret; + struct flb_azure_kusto *ctx; + + /* Allocate config context */ + ctx = flb_calloc(1, sizeof(struct flb_azure_kusto)); + if (!ctx) { + flb_errno(); + return NULL; + } + ctx->ins = ins; + ctx->config = config; + + ret = flb_output_config_map_set(ins, (void *)ctx); + if (ret == -1) { + flb_plg_error(ins, "unable to load configuration"); + flb_free(ctx); + return NULL; + } + + /* config: 'tenant_id' */ + if (ctx->tenant_id == NULL) { + flb_plg_error(ctx->ins, "property 'tenant_id' is not defined."); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + /* config: 'client_id' */ + if (ctx->client_id == NULL) { + flb_plg_error(ctx->ins, "property 'client_id' is not defined"); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + /* config: 'client_secret' */ + if (ctx->client_secret == NULL) { + flb_plg_error(ctx->ins, "property 'client_secret' is not defined"); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + /* config: 'ingestion_endpoint' */ + if (ctx->ingestion_endpoint == NULL) { + flb_plg_error(ctx->ins, "property 'ingestion_endpoint' is not defined"); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + /* config: 'database_name' */ + if (ctx->database_name == NULL) { + flb_plg_error(ctx->ins, "property 'database_name' is not defined"); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + /* config: 'table_name' */ + if (ctx->table_name == NULL) { + flb_plg_error(ctx->ins, "property 'table_name' is not defined"); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + /* Create the auth URL */ + ctx->oauth_url = flb_sds_create_size(sizeof(FLB_MSAL_AUTH_URL_TEMPLATE) - 1 + + flb_sds_len(ctx->tenant_id)); + if (!ctx->oauth_url) { + flb_errno(); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + flb_sds_snprintf(&ctx->oauth_url, flb_sds_alloc(ctx->oauth_url), + FLB_MSAL_AUTH_URL_TEMPLATE, ctx->tenant_id); + + ctx->resources = flb_calloc(1, sizeof(struct flb_azure_kusto_resources)); + if (!ctx->resources) { + flb_errno(); + flb_azure_kusto_conf_destroy(ctx); + return NULL; + } + + flb_plg_info(ctx->ins, "endpoint='%s', database='%s', table='%s'", + ctx->ingestion_endpoint, ctx->database_name, ctx->table_name); + + return ctx; +} + +int flb_azure_kusto_conf_destroy(struct flb_azure_kusto *ctx) +{ + if (!ctx) { + return -1; + } + + if (ctx->oauth_url) { + flb_sds_destroy(ctx->oauth_url); + ctx->oauth_url = NULL; + } + + if (ctx->o) { + flb_oauth2_destroy(ctx->o); + ctx->o = NULL; + } + + if (ctx->resources) { + flb_azure_kusto_resources_destroy(ctx->resources); + ctx->resources = NULL; + } + + flb_free(ctx); + return 0; +} diff --git a/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_conf.h b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_conf.h new file mode 100644 index 000000000..b4b2e3a39 --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_conf.h @@ -0,0 +1,31 @@ +/* -*- 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_AZURE_KUSTO_CONF_H +#define FLB_OUT_AZURE_KUSTO_CONF_H + +#include "azure_kusto.h" + +int azure_kusto_load_ingestion_resources(struct flb_azure_kusto *ctx, + struct flb_config *config); +struct flb_azure_kusto *flb_azure_kusto_conf_create(struct flb_output_instance *ins, + struct flb_config *config); +int flb_azure_kusto_conf_destroy(struct flb_azure_kusto *ctx); + +#endif diff --git a/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_ingest.c b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_ingest.c new file mode 100644 index 000000000..d38d92e7f --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_ingest.c @@ -0,0 +1,496 @@ +/* -*- 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_base64.h> +#include <fluent-bit/flb_http_client.h> +#include <fluent-bit/flb_output_plugin.h> +#include <fluent-bit/flb_random.h> +#include <fluent-bit/flb_time.h> +#include <fluent-bit/flb_utils.h> + +#include <math.h> +#include <msgpack.h> + +#include "azure_kusto_ingest.h" + +/* not really uuid but a random string in the form 00000000-0000-0000-0000-000000000000 */ +static char *generate_uuid() +{ + char *chars = "0123456789abcdef"; + char *uuid; + int i; + uint64_t rand; + + uuid = flb_malloc(37); + if (!uuid) { + flb_errno(); + return NULL; + } + + for (i = 0; i < 36; i++) { + if (i == 8 || i == 13 || i == 18 || i == 23) { + uuid[i] = '-'; + continue; + } + + if (flb_random_bytes((unsigned char *)&rand, sizeof(uint64_t))) { + rand = time(NULL); + } + uuid[i] = chars[rand % 16]; + } + uuid[36] = '\0'; + + return uuid; +} + +static char *base64_encode(flb_sds_t s, size_t len, size_t *out_len) +{ + char *b64; + int ret; + size_t buffer_len = 4 * ceil(((double)len / 3) + 1); + + b64 = flb_malloc(buffer_len); + if (!b64) { + flb_errno(); + return NULL; + } + + ret = flb_base64_encode((unsigned char *)b64, buffer_len, out_len, (unsigned char *)s, + len); + if (ret != 0) { + flb_error("cannot encode string %s into base64", s); + flb_free(b64); + return NULL; + } + + return b64; +} + +static flb_sds_t azure_kusto_create_blob_uri(struct flb_azure_kusto *ctx, + struct flb_upstream_node *u_node, + flb_sds_t blob_id) +{ + int ret; + flb_sds_t uri = NULL; + char *blob_uri; + size_t blob_uri_size; + char *blob_sas; + size_t blob_sas_size; + + ret = flb_hash_table_get(u_node->ht, AZURE_KUSTO_RESOURCE_UPSTREAM_URI, 3, + (void **)&blob_uri, &blob_uri_size); + if (ret == -1) { + flb_plg_error(ctx->ins, "error getting blob uri"); + return NULL; + } + + ret = flb_hash_table_get(u_node->ht, AZURE_KUSTO_RESOURCE_UPSTREAM_SAS, 3, + (void **)&blob_sas, &blob_sas_size); + if (ret == -1) { + flb_plg_error(ctx->ins, "error getting blob sas token"); + return NULL; + } + + /* uri will be https://<blob_host>/<container_uri>/<blob_id>.multijson?<sas_token> */ + uri = flb_sds_create_size(flb_sds_len(u_node->host) + blob_uri_size + blob_sas_size + + flb_sds_len(blob_id) + 21); + + if (uri) { + flb_sds_snprintf(&uri, flb_sds_alloc(uri), "https://%s%s/%s.multijson?%s", + u_node->host, blob_uri, blob_id, blob_sas); + flb_plg_debug(ctx->ins, "created blob uri %s", uri); + } + else { + flb_plg_error(ctx->ins, "cannot create blob uri buffer"); + } + + return uri; +} + +static flb_sds_t azure_kusto_create_blob(struct flb_azure_kusto *ctx, flb_sds_t blob_id, + flb_sds_t payload, size_t payload_size) +{ + int ret = -1; + flb_sds_t uri = NULL; + struct flb_upstream_node *u_node; + struct flb_connection *u_conn; + struct flb_http_client *c; + size_t resp_size; + time_t now; + struct tm tm; + char tmp[64]; + int len; + + now = time(NULL); + gmtime_r(&now, &tm); + len = strftime(tmp, sizeof(tmp) - 1, "%a, %d %b %Y %H:%M:%S GMT", &tm); + + u_node = flb_upstream_ha_node_get(ctx->resources->blob_ha); + if (!u_node) { + flb_plg_error(ctx->ins, "error getting blob upstream"); + return NULL; + } + + u_conn = flb_upstream_conn_get(u_node->u); + + if (u_conn) { + uri = azure_kusto_create_blob_uri(ctx, u_node, blob_id); + + if (uri) { + flb_plg_debug(ctx->ins, "uploading payload to blob uri: %s", uri); + c = flb_http_client(u_conn, FLB_HTTP_PUT, uri, payload, payload_size, NULL, 0, + NULL, 0); + + if (c) { + flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); + flb_http_add_header(c, "Content-Type", 12, "application/json", 16); + flb_http_add_header(c, "x-ms-blob-type", 14, "BlockBlob", 9); + flb_http_add_header(c, "x-ms-date", 9, tmp, len); + flb_http_add_header(c, "x-ms-version", 12, "2019-12-12", 10); + + ret = flb_http_do(c, &resp_size); + flb_plg_debug(ctx->ins, + "kusto blob upload request http_do=%i, HTTP Status: %i", + ret, c->resp.status); + + if (ret == 0) { + /* Validate return status and HTTP status if set */ + if (c->resp.status != 201) { + ret = -1; + + if (c->resp.payload_size > 0) { + flb_plg_debug(ctx->ins, "Request failed and returned: \n%s", + c->resp.payload); + } + else { + flb_plg_debug(ctx->ins, "Request failed"); + } + } + } + else { + flb_plg_error(ctx->ins, "cannot send HTTP request"); + } + + flb_http_client_destroy(c); + } + else { + flb_plg_error(ctx->ins, + "cannot create HTTP client context for blob container"); + } + + if (ret != 0) { + flb_sds_destroy(uri); + uri = NULL; + } + } + else { + flb_plg_error(ctx->ins, "error creating blob container uri buffer"); + } + + flb_upstream_conn_release(u_conn); + } + else { + flb_plg_error(ctx->ins, "error getting blob container upstream connection"); + } + + return uri; +} + +static flb_sds_t create_ingestion_message(struct flb_azure_kusto *ctx, flb_sds_t blob_uri, + size_t payload_size) +{ + flb_sds_t message = NULL; + int ret = 0; + char *uuid; + char *message_b64; + size_t b64_len; + size_t message_len; + + uuid = generate_uuid(); + if (uuid) { + message = flb_sds_create(NULL); + + if (message) { + message_len = + flb_sds_snprintf(&message, 0, + "{\"Id\": \"%s\", \"BlobPath\": \"%s\", " + "\"RawDataSize\": %lu, \"DatabaseName\": " + "\"%s\", \"TableName\": \"%s\"," + "\"AdditionalProperties\": { \"format\": \"multijson\", " + "\"authorizationContext\": " + "\"%s\", \"jsonMappingReference\": \"%s\" }}%c", + uuid, blob_uri, payload_size, ctx->database_name, + ctx->table_name, ctx->resources->identity_token, + ctx->ingestion_mapping_reference == NULL + ? "" + : ctx->ingestion_mapping_reference, 0); + + if (message_len != -1) { + flb_plg_debug(ctx->ins, "created ingestion message:\n%s", message); + message_b64 = base64_encode(message, message_len, &b64_len); + + if (message_b64) { + ret = flb_sds_snprintf( + &message, flb_sds_alloc(message), + "<QueueMessage><MessageText>%s</MessageText></QueueMessage>%c", + message_b64, 0); + + if (ret == -1) { + flb_plg_error(ctx->ins, "error creating ingestion queue message"); + } + + flb_free(message_b64); + } + else { + flb_plg_error(ctx->ins, "error encoding ingestion message to base64"); + } + } + else { + flb_plg_error(ctx->ins, "error creating ingestion message"); + ret = -1; + } + + if (ret == -1) { + flb_sds_destroy(message); + message = NULL; + } + } + else { + flb_plg_error(ctx->ins, "error creating ingestion message buffer"); + } + + flb_free(uuid); + } + else { + flb_plg_error(ctx->ins, "error generating unique ingestion UUID"); + } + + return message; +} + +static flb_sds_t azure_kusto_create_queue_uri(struct flb_azure_kusto *ctx, + struct flb_upstream_node *u_node) +{ + int ret; + flb_sds_t uri = NULL; + char *queue_uri; + size_t queue_uri_size; + char *queue_sas; + size_t queue_sas_size; + + ret = flb_hash_table_get(u_node->ht, AZURE_KUSTO_RESOURCE_UPSTREAM_URI, 3, + (void **)&queue_uri, &queue_uri_size); + if (ret == -1) { + flb_plg_error(ctx->ins, "error getting queue uri"); + return NULL; + } + + ret = flb_hash_table_get(u_node->ht, AZURE_KUSTO_RESOURCE_UPSTREAM_SAS, 3, + (void **)&queue_sas, &queue_sas_size); + if (ret == -1) { + flb_plg_error(ctx->ins, "error getting queue sas token"); + return NULL; + } + + /* uri will be <container_uri>/messages?<sas_token> */ + uri = flb_sds_create_size(queue_uri_size + queue_sas_size + 11); + + if (uri) { + flb_sds_snprintf(&uri, flb_sds_alloc(uri), "%s/messages?%s", queue_uri, + queue_sas); + flb_plg_debug(ctx->ins, "created queue uri %s", uri); + } + else { + flb_plg_error(ctx->ins, "cannot create queue uri buffer"); + } + + return uri; +} + +static int azure_kusto_enqueue_ingestion(struct flb_azure_kusto *ctx, flb_sds_t blob_uri, + size_t payload_size) +{ + int ret = -1; + struct flb_upstream_node *u_node; + struct flb_connection *u_conn; + struct flb_http_client *c; + flb_sds_t uri; + flb_sds_t payload; + size_t resp_size; + time_t now; + struct tm tm; + char tmp[64]; + int len; + + now = time(NULL); + gmtime_r(&now, &tm); + len = strftime(tmp, sizeof(tmp) - 1, "%a, %d %b %Y %H:%M:%S GMT", &tm); + + u_node = flb_upstream_ha_node_get(ctx->resources->queue_ha); + if (!u_node) { + flb_plg_error(ctx->ins, "error getting queue upstream"); + return -1; + } + + u_conn = flb_upstream_conn_get(u_node->u); + + if (u_conn) { + uri = azure_kusto_create_queue_uri(ctx, u_node); + + if (uri) { + payload = create_ingestion_message(ctx, blob_uri, payload_size); + + if (payload) { + c = flb_http_client(u_conn, FLB_HTTP_POST, uri, payload, + flb_sds_len(payload), NULL, 0, NULL, 0); + + if (c) { + flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); + flb_http_add_header(c, "Content-Type", 12, "application/atom+xml", + 20); + flb_http_add_header(c, "x-ms-date", 9, tmp, len); + flb_http_add_header(c, "x-ms-version", 12, "2019-12-12", 10); + + ret = flb_http_do(c, &resp_size); + flb_plg_debug(ctx->ins, + "kusto queue request http_do=%i, HTTP Status: %i", ret, + c->resp.status); + + if (ret == 0) { + /* Validate return status and HTTP status if set */ + if (c->resp.status != 201) { + ret = -1; + + if (c->resp.payload_size > 0) { + flb_plg_debug(ctx->ins, + "Request failed and returned: \n%s", + c->resp.payload); + } + else { + flb_plg_debug(ctx->ins, "Request failed"); + } + } + } + else { + flb_plg_error(ctx->ins, "cannot send HTTP request"); + } + + flb_http_client_destroy(c); + } + else { + flb_plg_error(ctx->ins, + "cannot create HTTP client context for queue"); + } + + flb_sds_destroy(payload); + } + else { + flb_plg_error(ctx->ins, "error creating payload buffer"); + } + + flb_sds_destroy(uri); + } + else { + flb_plg_error(ctx->ins, "error creating queue uri buffer"); + } + + flb_upstream_conn_release(u_conn); + } + else { + flb_plg_error(ctx->ins, "error getting queue upstream connection"); + } + + return ret; +} + +static flb_sds_t azure_kusto_create_blob_id(struct flb_azure_kusto *ctx, flb_sds_t tag, + size_t tag_len) +{ + flb_sds_t blob_id = NULL; + struct flb_time tm; + uint64_t ms; + char *b64tag; + size_t b64_len; + + flb_time_get(&tm); + ms = ((tm.tm.tv_sec * 1000) + (tm.tm.tv_nsec / 1000000)); + + b64tag = base64_encode(tag, tag_len, &b64_len); + + if (b64tag) { + /* remove trailing '=' */ + while (b64_len && b64tag[b64_len - 1] == '=') { + b64tag[b64_len - 1] = '\0'; + b64_len--; + } + + blob_id = flb_sds_create_size(flb_sds_len(ctx->database_name) + + flb_sds_len(ctx->table_name) + b64_len + 24); + if (blob_id) { + flb_sds_snprintf(&blob_id, flb_sds_alloc(blob_id), "flb__%s__%s__%s__%lu", + ctx->database_name, ctx->table_name, b64tag, ms); + } + else { + flb_plg_error(ctx->ins, "cannot create blob id buffer"); + } + + flb_free(b64tag); + } + else { + flb_plg_error(ctx->ins, "error encoding tag '%s' to base64", tag); + } + + return blob_id; +} + +int azure_kusto_queued_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag, + size_t tag_len, flb_sds_t payload, size_t payload_size) +{ + int ret = -1; + flb_sds_t blob_id; + flb_sds_t blob_uri; + + /* flb__<db>__<table>__<b64tag>__<timestamp> */ + blob_id = azure_kusto_create_blob_id(ctx, tag, tag_len); + + if (blob_id) { + blob_uri = azure_kusto_create_blob(ctx, blob_id, payload, payload_size); + + if (blob_uri) { + ret = azure_kusto_enqueue_ingestion(ctx, blob_uri, payload_size); + + if (ret != 0) { + flb_plg_error(ctx->ins, "failed to enqueue ingestion blob to queue"); + ret = -1; + } + + flb_sds_destroy(blob_uri); + } + else { + flb_plg_error(ctx->ins, "failed to create payload blob uri"); + } + + flb_sds_destroy(blob_id); + } + else { + flb_plg_error(ctx->ins, "cannot create blob id"); + } + + return ret; +} diff --git a/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_ingest.h b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_ingest.h new file mode 100644 index 000000000..60613919a --- /dev/null +++ b/src/fluent-bit/plugins/out_azure_kusto/azure_kusto_ingest.h @@ -0,0 +1,28 @@ +/* -*- 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_AZURE_KUSTO_INGEST_H +#define FLB_OUT_AZURE_KUSTO_INGEST_H + +#include "azure_kusto.h" + +int azure_kusto_queued_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag, + size_t tag_len, flb_sds_t payload, size_t payload_size); + +#endif
\ No newline at end of file |