out_splunk: add time_key support for the HEC event time

include/fluent-bit/flb_strptime.h

@@ -20,6 +20,8 @@
 #ifndef FLB_STRPTIME_H
 #define FLB_STRPTIME_H
 
+#include <fluent-bit/flb_time.h>
+
 char *flb_strptime(const char *s, const char *format, struct flb_tm *tm);
 
 #endif

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

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),

plugins/out_splunk/splunk.h

@@ -39,6 +39,7 @@
 #include <fluent-bit/flb_output_plugin.h>
 #include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_record_accessor.h>
+#include <fluent-bit/flb_time.h>
 
 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;
 

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