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/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); } 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; 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 } }; 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} };