Skip to content

Commit

Permalink
out_kinesis_firehose: used function flb_aws_strftime_precision for ti…
Browse files Browse the repository at this point in the history
…me output can support millisecond and nanosecond by adding %3N and %9N or %L to time_key_format.

Signed-off-by: Clay Cheng <[email protected]>
  • Loading branch information
Claych committed Nov 28, 2022
1 parent 9738ab9 commit 8ae0377
Showing 1 changed file with 20 additions and 8 deletions.
28 changes: 20 additions & 8 deletions plugins/out_kinesis_firehose/firehose_api.c
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,7 @@ static int process_event(struct flb_firehose *ctx, struct flush *buf,
size_t len;
size_t tmp_size;
void *compressed_tmp_buf;
char *out_buf;

tmp_buf_ptr = buf->tmp_buf + buf->tmp_buf_offset;
ret = flb_msgpack_to_json(tmp_buf_ptr,
Expand Down Expand Up @@ -216,14 +217,26 @@ static int process_event(struct flb_firehose *ctx, struct flush *buf,
ctx->delivery_stream);
return 2;
}
/* guess space needed to write time_key */
len = 6 + strlen(ctx->time_key) + 6 * strlen(ctx->time_key_format);

/* format time output and return the length */
len = flb_aws_strftime_precision(&out_buf, ctx->time_key_format, tms);
if (len == -1) {
/* when function flb_aws_strftime_precision's return value is -1,
* time_key will not be added to record.
*/
flb_plg_error(ctx->ins, "Failed to add time_key %s to record, %s",
ctx->time_key, ctx->delivery_stream);
return 0;
}

/* how much space do we have left */
tmp_size = (buf->tmp_buf_size - buf->tmp_buf_offset) - written;
if (len > tmp_size) {
/* not enough space- tell caller to retry */
/* not enough space - tell caller to retry */
flb_free(out_buf);
return 1;
}

time_key_ptr = tmp_buf_ptr + written - 1;
memcpy(time_key_ptr, ",", 1);
time_key_ptr++;
Expand All @@ -235,11 +248,10 @@ static int process_event(struct flb_firehose *ctx, struct flush *buf,
time_key_ptr += 3;
tmp_size = buf->tmp_buf_size - buf->tmp_buf_offset;
tmp_size -= (time_key_ptr - tmp_buf_ptr);
len = strftime(time_key_ptr, tmp_size, ctx->time_key_format, &time_stamp);
if (len <= 0) {
/* ran out of space - should not happen because of check above */
return 1;
}

/* merge out_buf to time_key_ptr */
memcpy(time_key_ptr, out_buf, len);
flb_free(out_buf);
time_key_ptr += len;
memcpy(time_key_ptr, "\"}", 2);
time_key_ptr += 2;
Expand Down

0 comments on commit 8ae0377

Please sign in to comment.