out_splunk: Add auto_extract_timestamp parameter

plugins/out_splunk/splunk.c

@@ -431,13 +431,15 @@ static int pack_map(struct flb_splunk *ctx, msgpack_packer *mp_pck,
     else {
         flb_mp_map_header_init(&mh, mp_pck);
 
-        /* Append the time key */
-        flb_mp_map_header_append(&mh);
-        msgpack_pack_str(mp_pck, sizeof(FLB_SPLUNK_DEFAULT_TIME) -1);
-        msgpack_pack_str_body(mp_pck,
-                              FLB_SPLUNK_DEFAULT_TIME,
-                              sizeof(FLB_SPLUNK_DEFAULT_TIME) - 1);
-        msgpack_pack_double(mp_pck, t);
+        if (ctx->auto_extract_timestamp == FLB_FALSE) {
+            /* Append the time key */
+            flb_mp_map_header_append(&mh);
+            msgpack_pack_str(mp_pck, sizeof(FLB_SPLUNK_DEFAULT_TIME) -1);
+            msgpack_pack_str_body(mp_pck,
+                                  FLB_SPLUNK_DEFAULT_TIME,
+                                  sizeof(FLB_SPLUNK_DEFAULT_TIME) - 1);
+            msgpack_pack_double(mp_pck, t);
+        }
 
         /* Pack Splunk metadata */
         pack_map_meta(ctx, &mh, mp_pck, map, tag, tag_len);
@@ -502,13 +504,15 @@ static inline int pack_event_key(struct flb_splunk *ctx, msgpack_packer *mp_pck,
     if (ctx->splunk_send_raw == FLB_FALSE) {
         flb_mp_map_header_init(&mh, mp_pck);
 
-        /* Append the time key */
-        flb_mp_map_header_append(&mh);
-        msgpack_pack_str(mp_pck, sizeof(FLB_SPLUNK_DEFAULT_TIME) -1);
-        msgpack_pack_str_body(mp_pck,
-                              FLB_SPLUNK_DEFAULT_TIME,
-                              sizeof(FLB_SPLUNK_DEFAULT_TIME) - 1);
-        msgpack_pack_double(mp_pck, t);
+        if (ctx->auto_extract_timestamp == FLB_FALSE) {
+            /* Append the time key */
+            flb_mp_map_header_append(&mh);
+            msgpack_pack_str(mp_pck, sizeof(FLB_SPLUNK_DEFAULT_TIME) -1);
+            msgpack_pack_str_body(mp_pck,
+                                  FLB_SPLUNK_DEFAULT_TIME,
+                                  sizeof(FLB_SPLUNK_DEFAULT_TIME) - 1);
+            msgpack_pack_double(mp_pck, t);
+        }
 
         /* Pack Splunk metadata */
         pack_map_meta(ctx, &mh, mp_pck, map, tag, tag_len);
@@ -869,6 +873,7 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
     flb_sds_t buf_data;
     size_t resp_size;
     size_t buf_size;
+    const char *endpoint;
     struct flb_splunk *ctx = out_context;
     struct flb_connection *u_conn;
     struct flb_http_client *c;
@@ -928,8 +933,13 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
         }
     }
 
+    endpoint = FLB_SPLUNK_DEFAULT_ENDPOINT;
+    if (ctx->auto_extract_timestamp == FLB_TRUE) {
+        endpoint = FLB_SPLUNK_AUTO_EXTRACT_ENDPOINT;
+    }
+
     /* Compose HTTP Client request */
-    c = flb_http_client(u_conn, FLB_HTTP_POST, FLB_SPLUNK_DEFAULT_ENDPOINT,
+    c = flb_http_client(u_conn, FLB_HTTP_POST, endpoint,
                         payload_buf, payload_size, NULL, 0, NULL, 0);
 
     /* HTTP Response buffer size, honor value set by the user */
@@ -1162,6 +1172,14 @@ static struct flb_config_map config_map[] = {
      "from the docs for more details to make this option work properly."
     },
 
+    {
+     FLB_CONFIG_MAP_BOOL, "auto_extract_timestamp", "off",
+     0, FLB_TRUE, offsetof(struct flb_splunk, auto_extract_timestamp),
+     "Ask Splunk to extract the timestamp from the event data by setting "
+     "auto_extract_timestamp=true in the HTTP Event Collector URL and omitting "
+     "the time field from the event envelope."
+    },
+
     {
      FLB_CONFIG_MAP_STR, "channel", NULL,
      0, FLB_TRUE, offsetof(struct flb_splunk, channel),

plugins/out_splunk/splunk.h

@@ -23,6 +23,8 @@
 #define FLB_SPLUNK_DEFAULT_HOST          "127.0.0.1"
 #define FLB_SPLUNK_DEFAULT_PORT          8088
 #define FLB_SPLUNK_DEFAULT_ENDPOINT      "/services/collector/event"
+#define FLB_SPLUNK_AUTO_EXTRACT_ENDPOINT \
+    "/services/collector/event?auto_extract_timestamp=true"
 #define FLB_SPLUNK_DEFAULT_TIME          "time"
 #define FLB_SPLUNK_DEFAULT_EVENT_HOST    "host"
 #define FLB_SPLUNK_DEFAULT_EVENT_SOURCE  "source"
@@ -108,6 +110,9 @@ struct flb_splunk {
     /* Send fields directly or pack data into "event" object */
     int splunk_send_raw;
 
+    /* Ask Splunk to extract the timestamp from the event data */
+    int auto_extract_timestamp;
+
     /* HTTP Client Setup */
     size_t buffer_size;
 

tests/integration/scenarios/out_splunk/config/out_splunk_auto_extract_timestamp.yaml

@@ -0,0 +1,19 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_splunk
+      dummy: '{"message":"2022-06-13 12:34:56 example event"}'
+      samples: 1
+  outputs:
+    - name: splunk
+      match: out_splunk
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      tls: off
+      splunk_token: secret-token
+      auto_extract_timestamp: true

tests/integration/scenarios/out_splunk/config/out_splunk_auto_extract_timestamp_event_key.yaml

@@ -0,0 +1,20 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_splunk
+      dummy: '{"message":"2022-06-13 12:34:56 example event"}'
+      samples: 1
+  outputs:
+    - name: splunk
+      match: out_splunk
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      tls: off
+      splunk_token: secret-token
+      event_key: $message
+      auto_extract_timestamp: true

tests/integration/scenarios/out_splunk/config/out_splunk_default.yaml

@@ -0,0 +1,18 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_splunk
+      dummy: '{"message":"2022-06-13 12:34:56 example event"}'
+      samples: 1
+  outputs:
+    - name: splunk
+      match: out_splunk
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      tls: off
+      splunk_token: secret-token

tests/integration/scenarios/out_splunk/tests/test_out_splunk_001.py

@@ -0,0 +1,94 @@
+import os
+
+import pytest
+import requests
+
+from server.http_server import configure_http_response, data_storage, http_server_run
+from utils.test_service import FluentBitTestService
+
+
+class Service:
+    def __init__(self, config_file):
+        self.config_file = os.path.abspath(
+            os.path.join(os.path.dirname(__file__), "../config", config_file)
+        )
+        self.service = FluentBitTestService(
+            self.config_file,
+            data_storage=data_storage,
+            data_keys=["payloads", "requests"],
+            pre_start=self._start_receiver,
+            post_stop=self._stop_receiver,
+        )
+
+    def _start_receiver(self, service):
+        http_server_run(service.test_suite_http_port)
+        configure_http_response(status_code=200, body={"text": "Success", "code": 0})
+        self.service.wait_for_http_endpoint(
+            f"http://127.0.0.1:{service.test_suite_http_port}/ping",
+            timeout=10,
+            interval=0.5,
+        )
+
+    def _stop_receiver(self, service):
+        try:
+            requests.post(
+                f"http://127.0.0.1:{service.test_suite_http_port}/shutdown",
+                timeout=2,
+            )
+        except requests.RequestException:
+            pass
+
+    def start(self):
+        self.service.start()
+
+    def stop(self):
+        self.service.stop()
+
+    def wait_for_requests(self, minimum_count, timeout=10):
+        if os.environ.get("VALGRIND"):
+            timeout = max(timeout * 3, 30)
+
+        return self.service.wait_for_condition(
+            lambda: data_storage["requests"]
+            if len(data_storage["requests"]) >= minimum_count
+            else None,
+            timeout=timeout,
+            interval=0.5,
+            description=f"{minimum_count} outbound Splunk requests",
+        )
+
+
+@pytest.mark.parametrize(
+    ("config_file", "expected_query", "expected_event_type"),
+    [
+        ("out_splunk_default.yaml", "", dict),
+        (
+            "out_splunk_auto_extract_timestamp.yaml",
+            "auto_extract_timestamp=true",
+            dict,
+        ),
+        (
+            "out_splunk_auto_extract_timestamp_event_key.yaml",
+            "auto_extract_timestamp=true",
+            str,
+        ),
+    ],
+    ids=["default", "auto_extract_timestamp", "auto_extract_timestamp_event_key"],
+)
+def test_out_splunk_auto_extract_timestamp(
+    config_file, expected_query, expected_event_type
+):
+    service = Service(config_file)
+    service.start()
+
+    try:
+        requests_seen = service.wait_for_requests(1)
+    finally:
+        service.stop()
+
+    first_request = requests_seen[0]
+    assert first_request["path"] == "/services/collector/event"
+    assert first_request["query_string"] == expected_query
+    assert first_request["headers"].get("Authorization") == "Splunk secret-token"
+    assert isinstance(first_request["json"]["event"], expected_event_type)
+    assert ("time" in first_request["json"]) == (expected_query == "")