From 6a1beee456702cd756bfcd7eeb457f67cb4bfdfc Mon Sep 17 00:00:00 2001 From: Anurag Gupta Date: Tue, 25 Aug 2026 09:43:30 -0700 Subject: [PATCH 1/4] time: add helpers to parse timestamps from record values Plugins that let users nominate a record key as the event time need to turn an arbitrary msgpack value into a timestamp, which means handling integers, floats, the event time extension and strings carrying either a numeric timestamp or a formatted one. That logic does not belong in a single plugin, so it is added to the core time API where it can be shared and covered by internal tests. flb_time_fmt_create() prepares a strptime(3) format once at initialization time. strptime(3) has no specifier for fractional seconds, so the format is split around '%L' and both halves are applied separately around the subsecond digits. flb_time_from_str() converts a string, and flb_time_from_msgpack_object() dispatches on the msgpack type. Values are rejected rather than silently accepted when they are not timestamps: a partial match leaving trailing data, and non finite floats, which cannot be represented as a timestamp and serialize to invalid JSON. A timezone offset matched before '%L' is preserved. flb_strptime() resets the offset on every call, so the part of the format that follows '%L' is parsed into a separate structure and only an offset it actually matched is carried over. The offset is also read before timegm(3) runs, since the conversion resets it on platforms where it is a member of 'struct tm'. flb_strptime.h is also made self contained. It referenced 'struct flb_tm' without declaring it, so the tag was created at function prototype scope and a file including it before flb_time.h failed to build with -Werror=incompatible-pointer-types. Signed-off-by: Anurag Gupta --- include/fluent-bit/flb_strptime.h | 2 + include/fluent-bit/flb_time.h | 20 +++ src/flb_time.c | 226 ++++++++++++++++++++++++++++++ 3 files changed, 248 insertions(+) diff --git a/include/fluent-bit/flb_strptime.h b/include/fluent-bit/flb_strptime.h index b788cdc41f7..d3eb8948c1b 100644 --- a/include/fluent-bit/flb_strptime.h +++ b/include/fluent-bit/flb_strptime.h @@ -20,6 +20,8 @@ #ifndef FLB_STRPTIME_H #define FLB_STRPTIME_H +#include + char *flb_strptime(const char *s, const char *format, struct flb_tm *tm); #endif diff --git a/include/fluent-bit/flb_time.h b/include/fluent-bit/flb_time.h index 440d887c374..b9102f1b092 100644 --- a/include/fluent-bit/flb_time.h +++ b/include/fluent-bit/flb_time.h @@ -53,6 +53,20 @@ struct flb_tm { #define flb_tm_zone(x) (x)->tm.tm_zone #endif +/* Maximum length of a timestamp string accepted by flb_time_from_str() */ +#define FLB_TIME_STR_MAX 64 + +/* + * Prepared strptime(3) format for timestamps that may carry fractional seconds + * through the '%L' specifier, which strptime(3) does not implement. The format + * is split around '%L' so both halves can be applied separately, before and + * after the subsecond digits. + */ +struct flb_time_fmt { + char *fmt; /* format up to '%L', or the whole format */ + char *frac_secs; /* remainder after '%L', NULL if '%L' is not present */ +}; + /* to represent eventtime of fluentd see also @@ -153,6 +167,12 @@ int flb_time_diff(struct flb_time *time1, int flb_time_append_to_mpack(mpack_writer_t *writer, struct flb_time *tm, int fmt); int flb_time_append_to_msgpack(struct flb_time *tm, msgpack_packer *pk, int fmt); int flb_time_msgpack_to_time(struct flb_time *time, msgpack_object *obj); +int flb_time_fmt_create(struct flb_time_fmt *tf, const char *format); +void flb_time_fmt_destroy(struct flb_time_fmt *tf); +int flb_time_from_str(struct flb_time *tm, const char *str, size_t len, + struct flb_time_fmt *tf); +int flb_time_from_msgpack_object(struct flb_time *tm, msgpack_object *obj, + struct flb_time_fmt *tf); int flb_time_pop_from_mpack(struct flb_time *time, mpack_reader_t *reader); int flb_time_pop_from_msgpack(struct flb_time *time, msgpack_unpacked *upk, msgpack_object **map); diff --git a/src/flb_time.c b/src/flb_time.c index 6e35c0dbb99..cce614c0215 100644 --- a/src/flb_time.c +++ b/src/flb_time.c @@ -23,6 +23,8 @@ #include #include #include +#include +#include #include #include #ifdef FLB_HAVE_CLOCK_GET_TIME @@ -30,6 +32,10 @@ # include #endif +#include +#include +#include +#include #include #include #include @@ -304,6 +310,226 @@ int flb_time_msgpack_to_time(struct flb_time *time, msgpack_object *obj) return 0; } +/* + * Parse the fractional seconds matched by the '%L' specifier: + * + * 2020-10-23T12:00:31.415213Z + * ------ + * + * Returns the number of characters consumed or -1 on error. + */ +static int parse_subseconds(const char *str, size_t len, double *subsec) +{ + int digits = 9; /* 1 ns = 000000001 (9 digits) */ + int consumed; + char *end; + char buf[16]; + + if (len < (size_t) digits) { + digits = (int) len; + } + + memcpy(buf, "0.", 2); + memcpy(buf + 2, str, digits); + buf[digits + 2] = '\0'; + + *subsec = strtod(buf, &end); + + consumed = end - buf - 2; + if (consumed <= 0) { + return -1; + } + + return consumed; +} + +int flb_time_fmt_create(struct flb_time_fmt *tf, const char *format) +{ + char *frac; + + if (tf == NULL || format == NULL) { + return -1; + } + + tf->frac_secs = NULL; + tf->fmt = flb_strdup(format); + if (tf->fmt == NULL) { + flb_errno(); + return -1; + } + + frac = strstr(tf->fmt, "%L"); + if (frac != NULL) { + *frac = '\0'; + tf->frac_secs = frac + 2; + } + + return 0; +} + +void flb_time_fmt_destroy(struct flb_time_fmt *tf) +{ + if (tf == NULL) { + return; + } + + if (tf->fmt != NULL) { + flb_free(tf->fmt); + tf->fmt = NULL; + } + + tf->frac_secs = NULL; +} + +/* + * Convert a timestamp string into 'tm'. When 'tf' holds a prepared format the + * value is parsed with strptime(3) semantics, otherwise the value is expected + * to contain a numeric Unix timestamp. + * + * The whole value must be consumed, a partial match is not considered a valid + * timestamp. + */ +int flb_time_from_str(struct flb_time *tm, const char *str, size_t len, + struct flb_time_fmt *tf) +{ + int consumed; + char *end; + char *p; + char buf[FLB_TIME_STR_MAX]; + long int gmtoff; + double subsec = 0.0; + double value; + struct tm tm_conv; + struct flb_tm tmp; + struct flb_tm frac_tmp; + + if (tm == NULL || str == NULL || len == 0 || len >= sizeof(buf)) { + return -1; + } + + /* both flb_strptime(3) and strtod(3) require a null terminated string */ + memcpy(buf, str, len); + buf[len] = '\0'; + + if (tf == NULL || tf->fmt == NULL) { + errno = 0; + value = strtod(buf, &end); + + /* + * non finite values are rejected: they cannot be represented as a + * timestamp and they serialize to invalid JSON. + */ + if (end == buf || errno == ERANGE || !isfinite(value)) { + return -1; + } + + while (isspace((unsigned char) *end)) { + end++; + } + + if (*end != '\0') { + return -1; + } + + tm->tm.tv_sec = (time_t) value; + tm->tm.tv_nsec = (long) ((value - (double) tm->tm.tv_sec) * + ONESEC_IN_NSEC); + + return 0; + } + + memset(&tmp, 0, sizeof(struct flb_tm)); + + p = flb_strptime(buf, tf->fmt, &tmp); + if (p == NULL) { + return -1; + } + + if (tf->frac_secs != NULL) { + consumed = parse_subseconds(p, len - (p - buf), &subsec); + if (consumed < 0) { + return -1; + } + p += consumed; + + /* + * flb_strptime() resets the timezone offset on every call, so the part + * of the format that follows '%L' is parsed into a separate structure + * and only a timezone that it actually matched is carried over. + */ + memset(&frac_tmp, 0, sizeof(struct flb_tm)); + + p = flb_strptime(p, tf->frac_secs, &frac_tmp); + if (p == NULL) { + return -1; + } + + if (flb_tm_gmtoff(&frac_tmp) != 0) { + flb_tm_gmtoff(&tmp) = flb_tm_gmtoff(&frac_tmp); + } + } + + while (isspace((unsigned char) *p)) { + p++; + } + + if (*p != '\0') { + return -1; + } + + /* + * timegm(3) normalizes the structure it receives, and on platforms where + * the timezone offset is a member of 'struct tm' it is reset by the + * conversion, so the offset is saved and a copy is handed over. + */ + gmtoff = flb_tm_gmtoff(&tmp); + tm_conv = tmp.tm; + + flb_time_set(tm, timegm(&tm_conv) - gmtoff, + (long) (subsec * ONESEC_IN_NSEC)); + + return 0; +} + +/* + * Extract a timestamp out of a record value. This extends + * flb_time_msgpack_to_time() with support for string values, which are parsed + * using 'tf', and it rejects non finite floats. + */ +int flb_time_from_msgpack_object(struct flb_time *tm, msgpack_object *obj, + struct flb_time_fmt *tf) +{ + if (tm == NULL || obj == NULL) { + return -1; + } + + switch (obj->type) { + case MSGPACK_OBJECT_POSITIVE_INTEGER: + flb_time_set(tm, (time_t) obj->via.u64, 0); + break; + case MSGPACK_OBJECT_NEGATIVE_INTEGER: + flb_time_set(tm, (time_t) obj->via.i64, 0); + break; + case MSGPACK_OBJECT_FLOAT32: + case MSGPACK_OBJECT_FLOAT64: + if (!isfinite(obj->via.f64)) { + return -1; + } + tm->tm.tv_sec = (time_t) obj->via.f64; + tm->tm.tv_nsec = (long) ((obj->via.f64 - (double) tm->tm.tv_sec) * + ONESEC_IN_NSEC); + break; + case MSGPACK_OBJECT_STR: + return flb_time_from_str(tm, obj->via.str.ptr, obj->via.str.size, tf); + case MSGPACK_OBJECT_EXT: + return flb_time_msgpack_to_time(tm, obj); + default: + return -1; + } + + return 0; +} + int flb_time_pop_from_mpack(struct flb_time *time, mpack_reader_t *reader) { mpack_tag_t tag; From 3bb6d3891fd8660b86dc2c25afd79b064985e42a Mon Sep 17 00:00:00 2001 From: Anurag Gupta Date: Tue, 25 Aug 2026 09:43:36 -0700 Subject: [PATCH 2/4] tests: internal: time: cover the timestamp parsing helpers Exercise flb_time_from_str() and flb_time_from_msgpack_object() for numeric strings, formatted strings with and without fractional seconds, nanosecond resolution and the msgpack types they accept. The rejected cases are covered as well, since falling back to the engine timestamp depends on them: values that do not match the format, trailing data after a complete match, '%L' with no digits to consume, non finite values and values longer than the accepted maximum. Two cases guard details that are easy to regress. A timezone offset placed before '%L' must survive the parsing of the fractional seconds, and it must still be applied after timegm(3) runs. Signed-off-by: Anurag Gupta --- tests/internal/flb_time.c | 328 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 328 insertions(+) diff --git a/tests/internal/flb_time.c b/tests/internal/flb_time.c index 3c0597196af..391e311ab0b 100644 --- a/tests/internal/flb_time.c +++ b/tests/internal/flb_time.c @@ -24,6 +24,8 @@ #include #include #include +#include +#include #include #include "flb_tests_internal.h" @@ -531,6 +533,323 @@ void test_iana_zone_to_utc_offset() } } +struct str_check { + const char *format; + const char *value; + time_t expect_sec; + long expect_nsec; +}; + +void test_from_str_numeric() +{ + int i; + int ret; + struct flb_time tm; + struct str_check checks[] = { + {NULL, "1647061992" , SEC_32BIT, 0}, + {NULL, "1647061992.123", SEC_32BIT, NSEC_32BIT}, + {NULL, " 1647061992 " , SEC_32BIT, 0}, + {NULL, NULL, 0, 0} + }; + + for (i = 0; checks[i].value != NULL; i++) { + ret = flb_time_from_str(&tm, checks[i].value, + strlen(checks[i].value), NULL); + if (!TEST_CHECK(ret == 0)) { + TEST_MSG("flb_time_from_str failed for '%s'", checks[i].value); + continue; + } + + if (!TEST_CHECK(tm.tm.tv_sec == checks[i].expect_sec && + labs(tm.tm.tv_nsec - checks[i].expect_nsec) < 10000)) { + TEST_MSG("value ='%s'", checks[i].value); + TEST_MSG("got =%ld.%ld", (long) tm.tm.tv_sec, tm.tm.tv_nsec); + TEST_MSG("expect =%ld.%ld", (long) checks[i].expect_sec, + checks[i].expect_nsec); + } + } +} + +void test_from_str_numeric_invalid() +{ + int i; + int ret; + struct flb_time tm; + const char *values[] = { + "", /* empty */ + "not-a-number", + "123abc", /* trailing garbage */ + "nan", /* non finite */ + "inf", + "infinity", + "-inf", + NULL + }; + + for (i = 0; values[i] != NULL; i++) { + ret = flb_time_from_str(&tm, values[i], strlen(values[i]), NULL); + if (!TEST_CHECK(ret != 0)) { + TEST_MSG("flb_time_from_str should fail for '%s', got %ld.%ld", + values[i], (long) tm.tm.tv_sec, tm.tm.tv_nsec); + } + } +} + +void test_from_str_format() +{ + int i; + int ret; + struct flb_time tm; + struct flb_time_fmt tf; + struct str_check checks[] = { + /* no fractional seconds */ + {"%Y-%m-%dT%H:%M:%S", "2022-03-12T05:13:12", SEC_32BIT, 0}, + /* '%L' at the end of the format */ + {"%Y-%m-%dT%H:%M:%S.%L", "2022-03-12T05:13:12.123", + SEC_32BIT, NSEC_32BIT}, + /* trailing literal after '%L' */ + {"%Y-%m-%dT%H:%M:%S.%LZ", "2022-03-12T05:13:12.123Z", + SEC_32BIT, NSEC_32BIT}, + /* timezone offset after '%L' */ + {"%Y-%m-%dT%H:%M:%S.%L%z", "2022-03-12T10:43:12.123+0530", + SEC_32BIT, NSEC_32BIT}, + /* + * timezone offset before '%L': flb_strptime() resets the offset on + * every call, so this checks that the offset parsed by the first pass + * survives the parsing of the fractional seconds. + */ + {"%Y-%m-%dT%H:%M:%S%z.%L", "2022-03-12T10:43:12+0530.123", + SEC_32BIT, NSEC_32BIT}, + /* nanosecond resolution */ + {"%Y-%m-%dT%H:%M:%S.%L", "2022-03-12T05:13:12.123456789", + SEC_32BIT, 123456789}, + {NULL, NULL, 0, 0} + }; + + for (i = 0; checks[i].value != NULL; i++) { + ret = flb_time_fmt_create(&tf, checks[i].format); + if (!TEST_CHECK(ret == 0)) { + TEST_MSG("flb_time_fmt_create failed for '%s'", checks[i].format); + continue; + } + + ret = flb_time_from_str(&tm, checks[i].value, + strlen(checks[i].value), &tf); + if (!TEST_CHECK(ret == 0)) { + TEST_MSG("flb_time_from_str failed for '%s' (format '%s')", + checks[i].value, checks[i].format); + flb_time_fmt_destroy(&tf); + continue; + } + + if (!TEST_CHECK(tm.tm.tv_sec == checks[i].expect_sec && + labs(tm.tm.tv_nsec - checks[i].expect_nsec) < 10000)) { + TEST_MSG("format ='%s'", checks[i].format); + TEST_MSG("value ='%s'", checks[i].value); + TEST_MSG("got =%ld.%ld", (long) tm.tm.tv_sec, tm.tm.tv_nsec); + TEST_MSG("expect =%ld.%ld", (long) checks[i].expect_sec, + checks[i].expect_nsec); + } + + flb_time_fmt_destroy(&tf); + } +} + +void test_from_str_format_invalid() +{ + int i; + int ret; + struct flb_time tm; + struct flb_time_fmt tf; + struct str_check checks[] = { + /* does not match the format at all */ + {"%Y-%m-%dT%H:%M:%S", "not a timestamp", 0, 0}, + /* trailing data after a complete match must be rejected */ + {"%Y-%m-%dT%H:%M:%SZ", "2022-03-12T05:13:12Zgarbage", 0, 0}, + {"%Y-%m-%dT%H:%M:%S.%L", "2022-03-12T05:13:12.123garbage", 0, 0}, + /* '%L' with no digits to consume */ + {"%Y-%m-%dT%H:%M:%S.%L", "2022-03-12T05:13:12.", 0, 0}, + {NULL, NULL, 0, 0} + }; + + for (i = 0; checks[i].value != NULL; i++) { + ret = flb_time_fmt_create(&tf, checks[i].format); + if (!TEST_CHECK(ret == 0)) { + TEST_MSG("flb_time_fmt_create failed for '%s'", checks[i].format); + continue; + } + + ret = flb_time_from_str(&tm, checks[i].value, + strlen(checks[i].value), &tf); + if (!TEST_CHECK(ret != 0)) { + TEST_MSG("flb_time_from_str should fail for '%s' (format '%s')", + checks[i].value, checks[i].format); + } + + flb_time_fmt_destroy(&tf); + } +} + +void test_from_str_too_long() +{ + int ret; + char value[FLB_TIME_STR_MAX + 8]; + struct flb_time tm; + + memset(value, '1', sizeof(value) - 1); + value[sizeof(value) - 1] = '\0'; + + ret = flb_time_from_str(&tm, value, strlen(value), NULL); + if (!TEST_CHECK(ret != 0)) { + TEST_MSG("flb_time_from_str should reject values longer than %d bytes", + FLB_TIME_STR_MAX); + } +} + +void test_fmt_create_invalid() +{ + struct flb_time_fmt tf; + + if (!TEST_CHECK(flb_time_fmt_create(&tf, NULL) != 0)) { + TEST_MSG("flb_time_fmt_create should fail on a NULL format"); + } + + if (!TEST_CHECK(flb_time_fmt_create(NULL, "%Y") != 0)) { + TEST_MSG("flb_time_fmt_create should fail on a NULL holder"); + } + + /* destroying a never created format must be safe */ + flb_time_fmt_destroy(NULL); +} + +void test_from_msgpack_object_str() +{ + int ret; + struct flb_time tm; + struct flb_time_fmt tf; + + msgpack_packer mp_pck; + msgpack_sbuffer mp_sbuf; + msgpack_unpacked result; + msgpack_object tm_obj; + + const char *value = "2022-03-12T05:13:12.123Z"; + + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + msgpack_pack_str_with_body(&mp_pck, value, strlen(value)); + + msgpack_unpacked_init(&result); + msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, NULL); + tm_obj = result.data; + + ret = flb_time_fmt_create(&tf, "%Y-%m-%dT%H:%M:%S.%LZ"); + TEST_CHECK(ret == 0); + + ret = flb_time_from_msgpack_object(&tm, &tm_obj, &tf); + if (!TEST_CHECK(ret == 0)) { + TEST_MSG("flb_time_from_msgpack_object failed"); + } + else if (!TEST_CHECK(tm.tm.tv_sec == SEC_32BIT && + labs(tm.tm.tv_nsec - NSEC_32BIT) < 10000)) { + TEST_MSG("got %ld.%ld, expect %d.%d", (long) tm.tm.tv_sec, + tm.tm.tv_nsec, SEC_32BIT, NSEC_32BIT); + } + + flb_time_fmt_destroy(&tf); + msgpack_sbuffer_destroy(&mp_sbuf); + msgpack_unpacked_destroy(&result); +} + +void test_from_msgpack_object_numbers() +{ + int ret; + struct flb_time tm; + + msgpack_packer mp_pck; + msgpack_sbuffer mp_sbuf; + msgpack_unpacked result; + msgpack_object tm_obj; + + /* positive integer */ + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + msgpack_pack_uint64(&mp_pck, SEC_32BIT); + msgpack_unpacked_init(&result); + msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, NULL); + tm_obj = result.data; + + ret = flb_time_from_msgpack_object(&tm, &tm_obj, NULL); + TEST_CHECK(ret == 0); + if (!TEST_CHECK(tm.tm.tv_sec == SEC_32BIT && tm.tm.tv_nsec == 0)) { + TEST_MSG("got %ld.%ld", (long) tm.tm.tv_sec, tm.tm.tv_nsec); + } + + msgpack_sbuffer_destroy(&mp_sbuf); + msgpack_unpacked_destroy(&result); + + /* negative integer, not handled by flb_time_msgpack_to_time() */ + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + msgpack_pack_int64(&mp_pck, -1); + msgpack_unpacked_init(&result); + msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, NULL); + tm_obj = result.data; + + ret = flb_time_from_msgpack_object(&tm, &tm_obj, NULL); + TEST_CHECK(ret == 0); + if (!TEST_CHECK(tm.tm.tv_sec == -1 && tm.tm.tv_nsec == 0)) { + TEST_MSG("got %ld.%ld", (long) tm.tm.tv_sec, tm.tm.tv_nsec); + } + + msgpack_sbuffer_destroy(&mp_sbuf); + msgpack_unpacked_destroy(&result); +} + +void test_from_msgpack_object_invalid() +{ + int ret; + struct flb_time tm; + + msgpack_packer mp_pck; + msgpack_sbuffer mp_sbuf; + msgpack_unpacked result; + msgpack_object tm_obj; + + /* a non finite float cannot be represented as a timestamp */ + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + msgpack_pack_double(&mp_pck, INFINITY); + msgpack_unpacked_init(&result); + msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, NULL); + tm_obj = result.data; + + ret = flb_time_from_msgpack_object(&tm, &tm_obj, NULL); + if (!TEST_CHECK(ret != 0)) { + TEST_MSG("flb_time_from_msgpack_object should reject a non finite " + "float"); + } + + msgpack_sbuffer_destroy(&mp_sbuf); + msgpack_unpacked_destroy(&result); + + /* an unsupported type must be rejected */ + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + msgpack_pack_true(&mp_pck); + msgpack_unpacked_init(&result); + msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, NULL); + tm_obj = result.data; + + ret = flb_time_from_msgpack_object(&tm, &tm_obj, NULL); + if (!TEST_CHECK(ret != 0)) { + TEST_MSG("flb_time_from_msgpack_object should reject a boolean"); + } + + msgpack_sbuffer_destroy(&mp_sbuf); + msgpack_unpacked_destroy(&result); +} + TEST_LIST = { { "flb_time_to_nanosec" , test_to_nanosec}, { "flb_time_append_to_mpack_v1" , test_append_to_mpack_v1}, @@ -545,5 +864,14 @@ TEST_LIST = { { "iana_zone_to_windows" , test_iana_zone_to_windows}, { "windows_zone_to_utc_offset" , test_windows_zone_to_utc_offset}, { "iana_zone_to_utc_offset" , test_iana_zone_to_utc_offset}, + { "from_str_numeric" , test_from_str_numeric}, + { "from_str_numeric_invalid" , test_from_str_numeric_invalid}, + { "from_str_format" , test_from_str_format}, + { "from_str_format_invalid" , test_from_str_format_invalid}, + { "from_str_too_long" , test_from_str_too_long}, + { "fmt_create_invalid" , test_fmt_create_invalid}, + { "from_msgpack_object_str" , test_from_msgpack_object_str}, + { "from_msgpack_object_numbers" , test_from_msgpack_object_numbers}, + { "from_msgpack_object_invalid" , test_from_msgpack_object_invalid}, { NULL, NULL } }; From 217cbc190802b4546b7a21902be0059b69d751a7 Mon Sep 17 00:00:00 2001 From: Anurag Gupta Date: Tue, 25 Aug 2026 09:43:45 -0700 Subject: [PATCH 3/4] out_splunk: add time_key support for the HEC event time The plugin always injected the Fluent Bit engine event time into the top level 'time' field of the HTTP Event Collector envelope, so a timestamp carried inside the record itself could not be reported to Splunk. Users had to either move to out_http and hand craft the envelope, or override _time on the indexer with custom timestamp extraction rules. Two new options are available now. 'time_key' names the record key that holds the event time and also accepts a record accessor pattern, while 'time_key_format' provides an optional strptime(3) format used when that key holds a string, including '%L' for fractional seconds. Integer, float, event time extension and numeric string values work without a format. Whenever the key is missing or its value cannot be parsed, the engine event time is used as before. The parsing itself is done by the core time API, so the plugin only resolves the record value and reports the outcome. On raw mode there is no HEC envelope to populate. The options are skipped entirely instead of being validated and then ignored, so an unrelated 'time_key' pattern cannot keep a raw mode output from starting. Signed-off-by: Anurag Gupta --- plugins/out_splunk/splunk.c | 63 ++++++++++++++++++++++++++++++- plugins/out_splunk/splunk.h | 9 +++++ plugins/out_splunk/splunk_conf.c | 65 ++++++++++++++++++++++++++++++++ 3 files changed, 135 insertions(+), 2 deletions(-) diff --git a/plugins/out_splunk/splunk.c b/plugins/out_splunk/splunk.c index 0881f8bf17f..1007733de8d 100644 --- a/plugins/out_splunk/splunk.c +++ b/plugins/out_splunk/splunk.c @@ -405,6 +405,44 @@ static int pack_map_meta(struct flb_splunk *ctx, return 0; } +/* + * Resolve the value for the top level HEC 'time' field. By default the Fluent + * Bit event timestamp is used, but when 'time_key' is set and the record holds + * a valid timestamp on that key, the record value takes precedence. + */ +static double get_event_time(struct flb_splunk *ctx, struct flb_time *tm, + msgpack_object map) +{ + int ret; + struct flb_time record_tm; + struct flb_ra_value *rval; + + if (ctx->ra_time_key == NULL) { + return flb_time_to_double(tm); + } + + rval = flb_ra_get_value_object(ctx->ra_time_key, map); + if (rval == NULL) { + flb_plg_debug(ctx->ins, + "time_key '%s' not found in record, using the event " + "timestamp", ctx->time_key); + return flb_time_to_double(tm); + } + + ret = flb_time_from_msgpack_object(&record_tm, &rval->o, + &ctx->time_key_fmt); + flb_ra_key_value_destroy(rval); + + if (ret != 0) { + flb_plg_warn(ctx->ins, + "could not parse a timestamp from time_key '%s', using " + "the event timestamp", ctx->time_key); + return flb_time_to_double(tm); + } + + return flb_time_to_double(&record_tm); +} + static int pack_map(struct flb_splunk *ctx, msgpack_packer *mp_pck, struct flb_time *tm, msgpack_object *group_metadata, @@ -422,13 +460,14 @@ static int pack_map(struct flb_splunk *ctx, msgpack_packer *mp_pck, msgpack_object v; struct flb_mp_map_header mh; - t = flb_time_to_double(tm); map_size = map.via.map.size; if (ctx->splunk_send_raw == FLB_TRUE) { msgpack_pack_map(mp_pck, map_size /* all k/v */); } else { + t = get_event_time(ctx, tm, map); + flb_mp_map_header_init(&mh, mp_pck); if (ctx->auto_extract_timestamp == FLB_FALSE) { @@ -491,7 +530,6 @@ static inline int pack_event_key(struct flb_splunk *ctx, msgpack_packer *mp_pck, struct flb_mp_map_header mh; flb_sds_t val; - t = flb_time_to_double(tm); val = flb_ra_translate(ctx->ra_event_key, tag, tag_len, map, NULL); if (!val || flb_sds_len(val) == 0) { if (val != NULL) { @@ -502,6 +540,8 @@ static inline int pack_event_key(struct flb_splunk *ctx, msgpack_packer *mp_pck, } if (ctx->splunk_send_raw == FLB_FALSE) { + t = get_event_time(ctx, tm, map); + flb_mp_map_header_init(&mh, mp_pck); if (ctx->auto_extract_timestamp == FLB_FALSE) { @@ -1150,6 +1190,25 @@ static struct flb_config_map config_map[] = { "it will have precedence over the value set in 'event_index'." }, + { + FLB_CONFIG_MAP_STR, "time_key", NULL, + 0, FLB_TRUE, offsetof(struct flb_splunk, time_key), + "Set a record key that will populate the top level 'time' field of the " + "HTTP Event Collector payload. A record accessor pattern is allowed, e.g: " + "'$aggregator_time'. If the key is not found or its value cannot be " + "interpreted as a timestamp, the Fluent Bit event timestamp is used " + "instead. This option is ignored when 'splunk_send_raw' is enabled." + }, + + { + FLB_CONFIG_MAP_STR, "time_key_format", NULL, + 0, FLB_TRUE, offsetof(struct flb_splunk, time_key_format), + "Set the strptime(3) compatible format used to parse the value of " + "'time_key' when it holds a string, e.g: '%Y-%m-%dT%H:%M:%S.%LZ'. The " + "'%L' specifier can be used for fractional seconds. If unset, string " + "values are expected to contain a numeric Unix timestamp." + }, + { FLB_CONFIG_MAP_SLIST_2, "event_field", NULL, FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct flb_splunk, event_fields), diff --git a/plugins/out_splunk/splunk.h b/plugins/out_splunk/splunk.h index 18e0cee9dc1..377dc379e5d 100644 --- a/plugins/out_splunk/splunk.h +++ b/plugins/out_splunk/splunk.h @@ -39,6 +39,7 @@ #include #include #include +#include struct flb_splunk_field { flb_sds_t key_name; @@ -88,6 +89,14 @@ struct flb_splunk { flb_sds_t event_index_key; struct flb_record_accessor *ra_event_index_key; + /* Event time: record key that holds the timestamp to report to Splunk */ + flb_sds_t time_key; + struct flb_record_accessor *ra_time_key; + + /* strptime(3) format used when the 'time_key' value is a string */ + flb_sds_t time_key_format; + struct flb_time_fmt time_key_fmt; + /* Event fields */ struct mk_list *event_fields; diff --git a/plugins/out_splunk/splunk_conf.c b/plugins/out_splunk/splunk_conf.c index 7e63c27d072..0136593aeac 100644 --- a/plugins/out_splunk/splunk_conf.c +++ b/plugins/out_splunk/splunk_conf.c @@ -90,6 +90,7 @@ struct flb_splunk *flb_splunk_conf_create(struct flb_output_instance *ins, int ret; int io_flags = 0; size_t size; + flb_sds_t pattern; flb_sds_t t; const char *tmp; struct flb_upstream *upstream; @@ -233,6 +234,64 @@ struct flb_splunk *flb_splunk_conf_create(struct flb_output_instance *ins, } } + /* + * Event time. On raw mode there is no HEC envelope to populate, so the + * option is skipped entirely instead of being validated and ignored. + * + * 'time_key' accepts a record accessor pattern, for convenience a plain + * record key is also accepted and promoted to a pattern. + */ + if (ctx->time_key && ctx->splunk_send_raw == FLB_TRUE) { + flb_plg_warn(ctx->ins, "'time_key' is ignored when 'splunk_send_raw' " + "is enabled"); + } + else if (ctx->time_key) { + if (ctx->time_key[0] == '$') { + ctx->ra_time_key = flb_ra_create(ctx->time_key, FLB_TRUE); + } + else { + pattern = flb_sds_create_size(flb_sds_len(ctx->time_key) + 1); + if (!pattern) { + flb_errno(); + flb_splunk_conf_destroy(ctx); + return NULL; + } + + t = flb_sds_printf(&pattern, "$%s", ctx->time_key); + if (!t) { + flb_errno(); + flb_sds_destroy(pattern); + flb_splunk_conf_destroy(ctx); + return NULL; + } + + ctx->ra_time_key = flb_ra_create(pattern, FLB_TRUE); + flb_sds_destroy(pattern); + } + + if (!ctx->ra_time_key) { + flb_plg_error(ctx->ins, + "cannot create record accessor for time_key " + "pattern: '%s'", ctx->time_key); + flb_splunk_conf_destroy(ctx); + return NULL; + } + } + + if (ctx->time_key_format && ctx->ra_time_key == NULL) { + flb_plg_warn(ctx->ins, "'time_key_format' has no effect because " + "'time_key' is not in use"); + } + else if (ctx->time_key_format) { + ret = flb_time_fmt_create(&ctx->time_key_fmt, ctx->time_key_format); + if (ret != 0) { + flb_plg_error(ctx->ins, "cannot prepare time_key_format '%s'", + ctx->time_key_format); + flb_splunk_conf_destroy(ctx); + return NULL; + } + } + /* Event fields */ ret = event_fields_create(ctx); if (ret == -1) { @@ -324,6 +383,12 @@ int flb_splunk_conf_destroy(struct flb_splunk *ctx) flb_ra_destroy(ctx->ra_event_index_key); } + if (ctx->ra_time_key) { + flb_ra_destroy(ctx->ra_time_key); + } + + flb_time_fmt_destroy(&ctx->time_key_fmt); + if (ctx->ra_metadata_auth_key) { flb_ra_destroy(ctx->ra_metadata_auth_key); } From 784576f8493d0b59eab00e88240869d04110740f Mon Sep 17 00:00:00 2001 From: Anurag Gupta Date: Tue, 25 Aug 2026 09:43:51 -0700 Subject: [PATCH 4/4] tests: runtime: out_splunk: add coverage for time_key Check that the HEC 'time' field is taken from the record for numeric values, for a record accessor pattern and for a formatted string, and that the engine timestamp is still used when the key is missing or holds a value that cannot be parsed. Also check that a 'time_key' configured together with 'splunk_send_raw' is ignored without keeping the output from starting. Signed-off-by: Anurag Gupta --- tests/runtime/out_splunk.c | 175 +++++++++++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) diff --git a/tests/runtime/out_splunk.c b/tests/runtime/out_splunk.c index 407c26b8b1c..e89a06b71de 100644 --- a/tests/runtime/out_splunk.c +++ b/tests/runtime/out_splunk.c @@ -142,9 +142,184 @@ void flb_test_basic() flb_destroy(ctx); } +#define JSON_TIME_NUM "[12345678, {\"key\":\"value\",\"event_time\":1700000000}]" +#define JSON_TIME_STR "[12345678, {\"key\":\"value\"," \ + "\"event_time\":\"2024-01-02T03:04:05.123Z\"}]" + +static void cb_check_time_key_num(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + char *p; + flb_sds_t out_js = res_data; + char *time_line = "\"time\":1700000000.0"; + + p = strstr(out_js, time_line); + if (!TEST_CHECK(p != NULL)) { + TEST_MSG("Given:%s", out_js); + } + + flb_sds_destroy(out_js); +} + +static void cb_check_time_key_str(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + char *p; + flb_sds_t out_js = res_data; + char *time_line = "\"time\":1704164645.123"; + + p = strstr(out_js, time_line); + if (!TEST_CHECK(p != NULL)) { + TEST_MSG("Given:%s", out_js); + } + + flb_sds_destroy(out_js); +} + +static void cb_check_time_key_fallback(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) +{ + char *p; + flb_sds_t out_js = res_data; + char *time_line = "\"time\":12345678.0"; + + p = strstr(out_js, time_line); + if (!TEST_CHECK(p != NULL)) { + TEST_MSG("Given:%s", out_js); + } + + flb_sds_destroy(out_js); +} + +static void flb_test_time_key_run(char *json, size_t json_size, + void (*cb)(void *, int, int, void *, size_t, + void *), + char *time_key, char *time_key_format) +{ + int ret; + flb_ctx_t *ctx; + int in_ffd; + int out_ffd; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "1", "grace", "1", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "splunk", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "http_user", "alice", + "time_key", time_key, + NULL); + + if (time_key_format) { + flb_output_set(ctx, out_ffd, "time_key_format", time_key_format, NULL); + } + + ret = flb_output_set_test(ctx, out_ffd, "formatter", cb, NULL, NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + flb_lib_push(ctx, in_ffd, json, json_size); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + +/* A numeric record key is used as the HEC event time */ +void flb_test_time_key_number() +{ + flb_test_time_key_run(JSON_TIME_NUM, sizeof(JSON_TIME_NUM) - 1, + cb_check_time_key_num, "event_time", NULL); +} + +/* A record accessor pattern is also a valid 'time_key' value */ +void flb_test_time_key_record_accessor() +{ + flb_test_time_key_run(JSON_TIME_NUM, sizeof(JSON_TIME_NUM) - 1, + cb_check_time_key_num, "$event_time", NULL); +} + +/* A string record key is parsed using 'time_key_format' */ +void flb_test_time_key_format() +{ + flb_test_time_key_run(JSON_TIME_STR, sizeof(JSON_TIME_STR) - 1, + cb_check_time_key_str, "event_time", + "%Y-%m-%dT%H:%M:%S.%LZ"); +} + +/* A missing 'time_key' falls back to the Fluent Bit event timestamp */ +void flb_test_time_key_missing() +{ + flb_test_time_key_run(JSON_BASIC, sizeof(JSON_BASIC) - 1, + cb_check_time_key_fallback, "event_time", NULL); +} + +/* An unparseable value falls back to the Fluent Bit event timestamp */ +void flb_test_time_key_invalid() +{ + flb_test_time_key_run(JSON_TIME_STR, sizeof(JSON_TIME_STR) - 1, + cb_check_time_key_fallback, "event_time", NULL); +} + +/* + * On raw mode there is no HEC envelope to populate, so 'time_key' must be + * ignored without preventing the output from starting. + */ +void flb_test_time_key_send_raw() +{ + int ret; + int size = sizeof(JSON_BASIC) - 1; + flb_ctx_t *ctx; + int in_ffd; + int out_ffd; + + ctx = flb_create(); + flb_service_set(ctx, "flush", "1", "grace", "1", NULL); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "splunk", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + "http_user", "alice", + "splunk_send_raw", "true", + "time_key", "event_time", + "time_key_format", "%Y-%m-%dT%H:%M:%S.%LZ", + NULL); + + ret = flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_send_raw, + NULL, NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + flb_lib_push(ctx, in_ffd, (char *) JSON_BASIC, size); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + /* Test list */ TEST_LIST = { {"basic" , flb_test_basic }, {"send_raw" , flb_test_send_raw}, + {"time_key_number" , flb_test_time_key_number}, + {"time_key_record_accessor", flb_test_time_key_record_accessor}, + {"time_key_format" , flb_test_time_key_format}, + {"time_key_missing" , flb_test_time_key_missing}, + {"time_key_invalid" , flb_test_time_key_invalid}, + {"time_key_send_raw", flb_test_time_key_send_raw}, {NULL, NULL} };