Skip to content

sds: general fixes for sds_cat() usage #11805

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 7 commits into from
May 15, 2026
Merged

sds: general fixes for sds_cat() usage #11805

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b1d418f
in_elasticsearch: check bulk status append failures
edsiper May 14, 2026
3a2a4f1
in_node_exporter_metrics: check meminfo append failure
edsiper May 14, 2026
a4b1d87
out_azure: check SDS append failures
edsiper May 14, 2026
9cc5bf6
out_azure_kusto: check SDS append failures
edsiper May 14, 2026
44787ed
out_bigquery: check JWT append failures
edsiper May 14, 2026
3e11246
out_stackdriver: check SDS append failures
edsiper May 14, 2026
f9c3dcc
tests: integration: accept curl send failure in bulk regression
edsiper May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 124 additions & 20 deletions plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,32 @@ static int status_buffer_avail(struct flb_in_elasticsearch *ctx, flb_sds_t bulk_
return FLB_TRUE;
}

static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char *buf, size_t size, flb_sds_t bulk_statuses)
static int bulk_statuses_cat(struct flb_in_elasticsearch *ctx,
flb_sds_t *bulk_statuses,
const char *data, size_t len)
{
flb_sds_t tmp;

if (flb_sds_avail(*bulk_statuses) < len) {
flb_plg_warn(ctx->ins, "left buffer for bulk status(es) is too small");

return FLB_FALSE;
}

tmp = flb_sds_cat(*bulk_statuses, data, len);
if (!tmp) {
flb_errno();

return FLB_FALSE;
}

*bulk_statuses = tmp;

return FLB_TRUE;
}

static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char *buf,
size_t size, flb_sds_t *bulk_statuses)
{
struct flb_log_event_encoder *encoder;
struct flb_log_event_encoder local_encoder;
Expand Down Expand Up @@ -146,29 +171,52 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
while (msgpack_unpack_next(&result, buf, size, &off) == MSGPACK_UNPACK_SUCCESS) {
if (result.data.type == MSGPACK_OBJECT_MAP) {
if (idx > 0 && idx % 2 == 0) {
flb_sds_cat(bulk_statuses, ",", 1);
if (status_buffer_avail(ctx, *bulk_statuses, 51) == FLB_FALSE) {
break;
}

if (bulk_statuses_cat(ctx, bulk_statuses, ",", 1) == FLB_FALSE) {
break;
}
}
if (status_buffer_avail(ctx, bulk_statuses, 50) == FLB_FALSE) {
else if (status_buffer_avail(ctx, *bulk_statuses, 50) == FLB_FALSE) {
break;
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
if (idx % 2 == 0) {
op_ret = get_write_op(ctx, &result.data, &write_op, &op_str_size);

if (op_ret) {
if (flb_sds_cmp(write_op, "index", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"index\":", 9);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"index\":", 9) == FLB_FALSE) {
flb_sds_destroy(write_op);
break;
}
error_op = FLB_FALSE;
}
else if (flb_sds_cmp(write_op, "create", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"create\":", 10);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"create\":", 10) == FLB_FALSE) {
flb_sds_destroy(write_op);
break;
}
error_op = FLB_FALSE;
}
else if (flb_sds_cmp(write_op, "update", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"update\":", 10);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"update\":", 10) == FLB_FALSE) {
flb_sds_destroy(write_op);
break;
}
error_op = FLB_TRUE;
}
else if (flb_sds_cmp(write_op, "delete", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"delete\":{\"status\":404,\"result\":\"not_found\"}}", 46);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"delete\":{\"status\":404,\"result\":\"not_found\"}}",
46) == FLB_FALSE) {
flb_sds_destroy(write_op);
break;
}
error_op = FLB_TRUE;
idx += 1; /* Prepare to adjust to multiple of two
* in the end of the loop.
Expand All @@ -178,7 +226,12 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
goto proceed;
}
else {
flb_sds_cat(bulk_statuses, "{\"unknown\":{\"status\":400,\"result\":\"bad_request\"}}", 49);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"unknown\":{\"status\":400,\"result\":\"bad_request\"}}",
49) == FLB_FALSE) {
flb_sds_destroy(write_op);
break;
}
error_op = FLB_TRUE;

flb_sds_destroy(write_op);
Expand Down Expand Up @@ -315,15 +368,36 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
}
if (op_ret) {
if (flb_sds_cmp(write_op, "index", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"status\":201,\"result\":\"created\"}}", 34);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"status\":201,\"result\":\"created\"}}",
34) == FLB_FALSE) {
flb_sds_destroy(write_op);
write_op = NULL;

break;
}
}
else if (flb_sds_cmp(write_op, "create", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"status\":201,\"result\":\"created\"}}", 34);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"status\":201,\"result\":\"created\"}}",
34) == FLB_FALSE) {
flb_sds_destroy(write_op);
write_op = NULL;

break;
}
}
else if (flb_sds_cmp(write_op, "update", op_str_size) == 0) {
flb_sds_cat(bulk_statuses, "{\"status\":403,\"result\":\"forbidden\"}}", 36);
if (bulk_statuses_cat(ctx, bulk_statuses,
"{\"status\":403,\"result\":\"forbidden\"}}",
36) == FLB_FALSE) {
flb_sds_destroy(write_op);
write_op = NULL;

break;
}
}
if (status_buffer_avail(ctx, bulk_statuses, 50) == FLB_FALSE) {
if (status_buffer_avail(ctx, *bulk_statuses, 50) == FLB_FALSE) {
flb_sds_destroy(write_op);
write_op = NULL;

Expand Down Expand Up @@ -383,7 +457,7 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
}

static ssize_t parse_payload_ndjson(struct flb_in_elasticsearch *ctx, flb_sds_t tag,
char *payload, size_t size, flb_sds_t bulk_statuses)
char *payload, size_t size, flb_sds_t *bulk_statuses)
{
int ret;
int out_size;
Expand Down Expand Up @@ -529,7 +603,7 @@ static int process_payload_ng(struct flb_http_request *request,
struct flb_http_response *response,
struct flb_in_elasticsearch *context,
flb_sds_t tag,
flb_sds_t bulk_statuses)
flb_sds_t *bulk_statuses)
{
if (request->content_type == NULL) {
send_response_ng(response, 400, NULL, "error: header 'Content-Type' is not set\n");
Expand Down Expand Up @@ -563,6 +637,7 @@ int in_elasticsearch_bulk_prot_handle_ng(struct flb_http_request *request,
int result;
flb_sds_t tag;
size_t len;
flb_sds_t tmp;

bulk_statuses = NULL;
bulk_response = NULL;
Expand Down Expand Up @@ -627,7 +702,7 @@ int in_elasticsearch_bulk_prot_handle_ng(struct flb_http_request *request,
return -1;
}

result = process_payload_ng(request, response, context, tag, bulk_statuses);
result = process_payload_ng(request, response, context, tag, &bulk_statuses);

flb_sds_destroy(tag);

Expand All @@ -647,20 +722,49 @@ int in_elasticsearch_bulk_prot_handle_ng(struct flb_http_request *request,
len = flb_sds_len(bulk_statuses);

if (flb_sds_alloc(bulk_response) < len + 27) {
bulk_response = flb_sds_increase(bulk_response, len + 27 - flb_sds_alloc(bulk_response));
tmp = flb_sds_increase(bulk_response, len + 27 - flb_sds_alloc(bulk_response));

if (tmp == NULL) {
flb_sds_destroy(bulk_statuses);
flb_sds_destroy(bulk_response);
return -1;
}

bulk_response = tmp;
}

error_str = strstr(bulk_statuses, "\"status\":40");

if (error_str){
flb_sds_cat(bulk_response, "{\"errors\":true,\"items\":[", 24);
tmp = flb_sds_cat(bulk_response, "{\"errors\":true,\"items\":[", 24);
}
else {
flb_sds_cat(bulk_response, "{\"errors\":false,\"items\":[", 25);
tmp = flb_sds_cat(bulk_response, "{\"errors\":false,\"items\":[", 25);
}

if (tmp == NULL) {
flb_sds_destroy(bulk_statuses);
flb_sds_destroy(bulk_response);
return -1;
}

bulk_response = tmp;

tmp = flb_sds_cat(bulk_response, bulk_statuses, flb_sds_len(bulk_statuses));
if (tmp == NULL) {
flb_sds_destroy(bulk_statuses);
flb_sds_destroy(bulk_response);
return -1;
}
bulk_response = tmp;

flb_sds_cat(bulk_response, bulk_statuses, flb_sds_len(bulk_statuses));
flb_sds_cat(bulk_response, "]}", 2);
tmp = flb_sds_cat(bulk_response, "]}", 2);
if (tmp == NULL) {
flb_sds_destroy(bulk_statuses);
flb_sds_destroy(bulk_response);
return -1;
}
bulk_response = tmp;

send_json_response_ng(response, 200, bulk_response);

Expand Down
30 changes: 27 additions & 3 deletions plugins/in_node_exporter_metrics/ne_meminfo_linux.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,11 @@ static int meminfo_configure(struct flb_ne *ctx)
entry->str[len] = '\0';

flb_sds_len_set(metric_name, 0);
flb_sds_cat(metric_name, entry->str, flb_sds_len(entry->str));
ret = flb_sds_cat_safe(&metric_name, entry->str, flb_sds_len(entry->str));
if (ret != 0) {
flb_slist_destroy(&split_list);
goto error;
}

/* Metric description */
flb_sds_len_set(metric_desc, 0);
Expand Down Expand Up @@ -254,8 +258,28 @@ static int meminfo_update(struct flb_ne *ctx)
else if (parts == 3) {
/* Compose new metric name */
tmp = flb_sds_create_size(256);
flb_sds_cat_safe(&tmp, metric_name, flb_sds_len(metric_name) - 1);
flb_sds_cat_safe(&tmp, "_bytes", 6);
if (!tmp) {
flb_errno();
flb_slist_destroy(&split_list);
flb_slist_destroy(&list);
return -1;
}

ret = flb_sds_cat_safe(&tmp, metric_name, flb_sds_len(metric_name) - 1);
if (ret == -1) {
flb_sds_destroy(tmp);
flb_slist_destroy(&split_list);
flb_slist_destroy(&list);
return -1;
}

ret = flb_sds_cat_safe(&tmp, "_bytes", 6);
if (ret == -1) {
flb_sds_destroy(tmp);
flb_slist_destroy(&split_list);
flb_slist_destroy(&list);
return -1;
}

/* Get metric context */
ret = flb_hash_table_get(ctx->meminfo_ht,
Expand Down
56 changes: 49 additions & 7 deletions plugins/out_azure/azure.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,7 @@ static int build_headers(struct flb_http_client *c,
size_t olen;
flb_sds_t rfc1123date;
flb_sds_t str_hash;
flb_sds_t tmp_sds;
struct tm tm = {0};
unsigned char hmac_hash[32] = {0};
int result;
Expand Down Expand Up @@ -231,13 +232,48 @@ static int build_headers(struct flb_http_client *c,
}

len = snprintf(tmp, sizeof(tmp) - 1, "%zu\n", content_length);
flb_sds_cat(str_hash, "POST\n", 5);
flb_sds_cat(str_hash, tmp, len);
flb_sds_cat(str_hash, "application/json\n", 17);
flb_sds_cat(str_hash, "x-ms-date:", 10);
flb_sds_cat(str_hash, rfc1123date, flb_sds_len(rfc1123date));
flb_sds_cat(str_hash, "\n", 1);
flb_sds_cat(str_hash, FLB_AZURE_RESOURCE, sizeof(FLB_AZURE_RESOURCE) - 1);
tmp_sds = flb_sds_cat(str_hash, "POST\n", 5);
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

tmp_sds = flb_sds_cat(str_hash, tmp, len);
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

tmp_sds = flb_sds_cat(str_hash, "application/json\n", 17);
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

tmp_sds = flb_sds_cat(str_hash, "x-ms-date:", 10);
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

tmp_sds = flb_sds_cat(str_hash, rfc1123date, flb_sds_len(rfc1123date));
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

tmp_sds = flb_sds_cat(str_hash, "\n", 1);
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

tmp_sds = flb_sds_cat(str_hash, FLB_AZURE_RESOURCE,
sizeof(FLB_AZURE_RESOURCE) - 1);
if (!tmp_sds) {
goto concat_error;
}
str_hash = tmp_sds;

/* Authorization signature */
result = flb_hmac_simple(FLB_HASH_SHA256,
Expand Down Expand Up @@ -291,6 +327,12 @@ static int build_headers(struct flb_http_client *c,
flb_free(auth);

return 0;

concat_error:
flb_sds_destroy(rfc1123date);
flb_sds_destroy(str_hash);

return -1;
}

static void cb_azure_flush(struct flb_event_chunk *event_chunk,
Expand Down
Loading
Loading