Feat: Add auto_extract_timestamp parameter

Ludovic Boutros · Ludovic Boutros · commit 4e58b43b6e2a · 2023-10-18T15:12:49.000+02:00
diff --git a/src/main/java/com/splunk/hecclient/HecConfig.java b/src/main/java/com/splunk/hecclient/HecConfig.java
@@ -40,6 +40,7 @@ public final class HecConfig {
     private String kerberosPrincipal;
     private String kerberosKeytabPath;
     private int concurrentHecQueueCapacity = 100;
+    private Boolean autoExtractTimestamp;
 
     public HecConfig(List<String> uris, String token) {
         this.uris = uris;
@@ -114,6 +115,8 @@ public int getConcurrentHecQueueCapacity() {
 
     public String getTrustStorePassword() { return trustStorePassword; }
 
+    public Boolean getAutoExtractTimestamp() { return autoExtractTimestamp; }
+
     public HecConfig setDisableSSLCertVerification(boolean disableVerfication) {
         disableSSLCertVerification = disableVerfication;
         return this;
@@ -217,6 +220,11 @@ public HecConfig setConcurrentHecQueueCapacity(int concurrentHecQueueCapacity) {
         return this;
     }
 
+    public HecConfig setAutoExtractTimestamp(Boolean autoExtractTimestamp) {
+        this.autoExtractTimestamp = autoExtractTimestamp;
+        return this;
+    }
+
     public boolean kerberosAuthEnabled() {
         return !kerberosPrincipal().isEmpty();
     }
diff --git a/src/main/java/com/splunk/hecclient/HecURIBuilder.java b/src/main/java/com/splunk/hecclient/HecURIBuilder.java
@@ -0,0 +1,32 @@
+package com.splunk.hecclient;
+
+import org.apache.http.client.utils.URIBuilder;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+public class HecURIBuilder {
+    public static final String AUTO_EXTRACT_TIMESTAMP_PARAMETER = "auto_extract_timestamp";
+
+    private final String baseUrl;
+    private final HecConfig hecConfig;
+
+    public HecURIBuilder(String baseUrl, HecConfig hecConfig) {
+        this.baseUrl = baseUrl;
+        this.hecConfig = hecConfig;
+    }
+
+    public URI getURI(String endpoint) {
+        try {
+            URIBuilder uriBuilder = new URIBuilder(baseUrl)
+                    .setPath(endpoint);
+
+            if (hecConfig.getAutoExtractTimestamp() != null) {
+                uriBuilder.addParameter(AUTO_EXTRACT_TIMESTAMP_PARAMETER, hecConfig.getAutoExtractTimestamp().toString());
+            }
+             return  uriBuilder.build();
+        } catch (URISyntaxException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}
diff --git a/src/main/java/com/splunk/hecclient/Indexer.java b/src/main/java/com/splunk/hecclient/Indexer.java
@@ -20,6 +20,9 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.splunk.kafka.connect.VersionUtils;
 import com.sun.security.auth.module.Krb5LoginModule;
+
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.security.Principal;
 import java.security.PrivilegedAction;
 import java.util.HashMap;
@@ -37,6 +40,7 @@
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.message.BasicHeader;
 import org.apache.http.protocol.HttpContext;
@@ -51,6 +55,8 @@ final class Indexer implements IndexerInf {
     private static final ObjectMapper jsonMapper = new ObjectMapper();
 
     private HecConfig hecConfig;
+
+    private HecURIBuilder hecURIBuilder;
     private Configuration config;
     private CloseableHttpClient httpClient;
     private HttpContext context;
@@ -72,6 +78,7 @@ public Indexer(String baseUrl,CloseableHttpClient client,Poller poller,HecConfig
         this.hecToken = config.getToken();
         this.poller = poller;
         this.context = HttpClientContext.create();
+        this.hecURIBuilder = new HecURIBuilder(baseUrl, hecConfig);
         backPressure = 0;
 
         channel = new HecChannel(this);
@@ -132,8 +139,8 @@ public HecChannel getChannel() {
     @Override
     public boolean send(final EventBatch batch) {
         String endpoint = batch.getRestEndpoint();
-        String url = baseUrl + endpoint;
-        final HttpPost httpPost = new HttpPost(url);
+        URI uri = hecURIBuilder.getURI(endpoint);
+        final HttpPost httpPost = new HttpPost(uri);
         httpPost.setHeaders(headers);
         if (batch.isEnableCompression()) {
             httpPost.setHeader("Content-Encoding", "gzip");
diff --git a/src/main/java/com/splunk/kafka/connect/SplunkSinkConnectorConfig.java b/src/main/java/com/splunk/kafka/connect/SplunkSinkConnectorConfig.java
@@ -80,6 +80,8 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
     static final String SSL_TRUSTSTORE_PATH_CONF = "splunk.hec.ssl.trust.store.path";
     static final String SSL_TRUSTSTORE_TYPE_CONF = "splunk.hec.ssl.trust.store.type";
     static final String SSL_TRUSTSTORE_PASSWORD_CONF = "splunk.hec.ssl.trust.store.password";
+    static final String AUTO_EXTRACT_TIMESTAMP_CONF = "splunk.hec.auto.extract.timestamp";
+
     //Headers
     static final String HEADER_SUPPORT_CONF = "splunk.header.support";
     static final String HEADER_CUSTOM_CONF = "splunk.header.custom";
@@ -187,6 +189,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
     static final String SSL_TRUSTSTORE_PATH_DOC = "Path on the local disk to the certificate trust store.";
     static final String SSL_TRUSTSTORE_TYPE_DOC = "Type of the trust store (JKS, PKCS12, ...).";
     static final String SSL_TRUSTSTORE_PASSWORD_DOC = "Password for the trust store.";
+    static final String AUTO_EXTRACT_TIMESTAMP_DOC = "Sends timestamped events to HTTP Event Collector using the Splunk platform JSON event protocol when auto_extract_timestamp is set to \"true\" in the /event URL.";
 
     static final String HEADER_SUPPORT_DOC = "Setting will enable Kafka Record headers to be used for meta data override";
     static final String HEADER_CUSTOM_DOC = "Setting will enable look for Record headers with these values and add them"
@@ -264,8 +267,8 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
     final String regex;
     final String timestampFormat;
     final int queueCapacity;
-
     final String timeZone;
+    final Boolean autoExtractTimestamp;
 
     SplunkSinkConnectorConfig(Map<String, String> taskConfig) {
         super(conf(), taskConfig);
@@ -324,6 +327,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         validateRegexForTimestamp(regex);
         queueCapacity = getInt(QUEUE_CAPACITY_CONF);
         validateQueueCapacity(queueCapacity);
+        autoExtractTimestamp = getBoolean(AUTO_EXTRACT_TIMESTAMP_CONF);
     }
 
    
@@ -341,6 +345,7 @@ public static ConfigDef conf() {
                 .define(SSL_TRUSTSTORE_PATH_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.HIGH, SSL_TRUSTSTORE_PATH_DOC)
                 .define(SSL_TRUSTSTORE_TYPE_CONF, ConfigDef.Type.STRING, "JKS", ConfigDef.Importance.LOW, SSL_TRUSTSTORE_TYPE_DOC)
                 .define(SSL_TRUSTSTORE_PASSWORD_CONF, ConfigDef.Type.PASSWORD, "", ConfigDef.Importance.HIGH, SSL_TRUSTSTORE_PASSWORD_DOC)
+                .define(AUTO_EXTRACT_TIMESTAMP_CONF, ConfigDef.Type.BOOLEAN, null, ConfigDef.Importance.LOW, AUTO_EXTRACT_TIMESTAMP_DOC)
                 .define(EVENT_TIMEOUT_CONF, ConfigDef.Type.INT, 300, ConfigDef.Importance.MEDIUM, EVENT_TIMEOUT_DOC)
                 .define(ACK_POLL_INTERVAL_CONF, ConfigDef.Type.INT, 10, ConfigDef.Importance.MEDIUM, ACK_POLL_INTERVAL_DOC)
                 .define(ACK_POLL_THREADS_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.MEDIUM, ACK_POLL_THREADS_DOC)
@@ -398,7 +403,8 @@ public HecConfig getHecConfig() {
               .setHasCustomTrustStore(hasTrustStorePath)
               .setKerberosPrincipal(kerberosUserPrincipal)
               .setKerberosKeytabPath(kerberosKeytabPath)
-              .setConcurrentHecQueueCapacity(queueCapacity);
+              .setConcurrentHecQueueCapacity(queueCapacity)
+              .setAutoExtractTimestamp(autoExtractTimestamp);
         return config;
     }
 
diff --git a/src/test/java/com/splunk/hecclient/HecURIBuilderTest.java b/src/test/java/com/splunk/hecclient/HecURIBuilderTest.java
@@ -0,0 +1,50 @@
+package com.splunk.hecclient;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.net.URI;
+import java.util.Collections;
+
+import static com.splunk.hecclient.JsonEventBatch.ENDPOINT;
+
+public class HecURIBuilderTest {
+    private static final String BASE_URL =  "https://localhost:8088";
+    private static final String TOKEN =  "mytoken";
+
+    @Test
+    public void testDefaultValues() {
+        HecConfig hecConfig = new HecConfig(Collections.emptyList(), TOKEN);
+        HecURIBuilder builder = new HecURIBuilder(BASE_URL, hecConfig);
+
+        URI uri = builder.getURI(ENDPOINT);
+
+        Assert.assertEquals("https://localhost:8088/services/collector/event", uri.toString());
+    }
+
+    @Test
+    public void testAutoExtractTimestamp() {
+        {
+            HecConfig hecConfig = new HecConfig(Collections.emptyList(), TOKEN)
+                    .setAutoExtractTimestamp(true);
+            HecURIBuilder builder = new HecURIBuilder(BASE_URL, hecConfig);
+
+            URI uri = builder.getURI(ENDPOINT);
+
+            Assert.assertEquals("https://localhost:8088/services/collector/event?" +
+                            HecURIBuilder.AUTO_EXTRACT_TIMESTAMP_PARAMETER + "=true",
+                    uri.toString());
+        }
+        {
+            HecConfig hecConfig = new HecConfig(Collections.emptyList(), TOKEN)
+                    .setAutoExtractTimestamp(false);
+            HecURIBuilder builder = new HecURIBuilder(BASE_URL, hecConfig);
+
+            URI uri = builder.getURI(ENDPOINT);
+
+            Assert.assertEquals("https://localhost:8088/services/collector/event?" +
+                            HecURIBuilder.AUTO_EXTRACT_TIMESTAMP_PARAMETER + "=false",
+                    uri.toString());
+        }
+    }
+}
