SUMOK-18 Added transformer into Properties file

Author: Juan Pablo Diaz-Vaz

src/main/java/com/mcplusa/kinesis/KinesisConnectorExecutor.java

@@ -38,9 +38,11 @@ public abstract class KinesisConnectorExecutor<T, U> extends KinesisConnectorExe
     // Create Stream Source constants
     private static final String CREATE_STREAM_SOURCE = "createStreamSource";
     private static final String LOOP_OVER_STREAM_SOURCE = "loopOverStreamSource";
+    private static final String INPUT_STREAM_FILE = "inputStreamFile";
+    
     private static final boolean DEFAULT_CREATE_STREAM_SOURCE = false;
     private static final boolean DEFAULT_LOOP_OVER_STREAM_SOURCE = false;
-    private static final String INPUT_STREAM_FILE = "inputStreamFile";
+
 
     // Class variables
     protected final KinesisConnectorForSumologicConfiguration config;

src/main/java/com/mcplusa/sumologic/DefaultKinesisMessageModelSumologicTransformer.java

@@ -0,0 +1,52 @@
+package com.mcplusa.sumologic;
+
+import java.io.IOException;
+
+import com.amazonaws.services.kinesis.connectors.BasicJsonTransformer;
+import com.amazonaws.services.kinesis.model.Record;
+import com.mcplusa.sumologic.KinesisMessageModel;
+import com.mcplusa.sumologic.implementations.SumologicEmitter;
+import com.mcplusa.sumologic.implementations.SumologicTransformer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.util.zip.GZIPInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import org.apache.commons.codec.binary.Base64;
+
+
+/**
+ * A custom transfomer for {@link KinesisMessageModel} records in JSON format. The output is in a format
+ * usable for insertions to Sumologic.
+ */
+public class DefaultKinesisMessageModelSumologicTransformer implements
+        SumologicTransformer<KinesisMessageModel> {
+
+  private static final Log LOG = LogFactory.getLog(DefaultKinesisMessageModelSumologicTransformer.class);
+  
+    /**
+     * Creates a new KinesisMessageModelSumologicTransformer.
+     */
+    public DefaultKinesisMessageModelSumologicTransformer() {
+        super();
+    }
+
+    @Override
+    public String fromClass(KinesisMessageModel message) {
+        return message.toString();
+    }
+
+    @Override
+    public KinesisMessageModel toClass(Record record) throws IOException {
+      byte[] decodedRecord = record.getData().array();
+      String stringifiedRecord = new String(decodedRecord);
+
+      return new KinesisMessageModel(stringifiedRecord);
+    }
+}

src/main/java/com/mcplusa/sumologic/KinesisConnectorForSumologicConfiguration.java

@@ -5,7 +5,6 @@
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.services.kinesis.connectors.KinesisConnectorConfiguration;
 
-
 /**
  * This class contains constants used to configure AWS Services in Amazon Kinesis Connectors. The user
  * should use System properties to set their proper configuration. An instance of
@@ -14,25 +13,22 @@
 public class KinesisConnectorForSumologicConfiguration extends KinesisConnectorConfiguration {
     // Properties added for Sumologic
     public static final String PROP_SUMOLOGIC_URL = "sumologicUrl";
-    public static final String PROP_SUMOLOGIC_USE_LOG4J = "useLog4j";
-    public final String SUMOLOGIC_URL;
-    public final boolean SUMOLOGIC_USE_LOG4J;
+    public static final String PROP_TRANSFORMER_CLASS = "transformerClass";
 
-    public final boolean DEFAULT_SUMOLOGIC_USE_LOG4J = false;
+    private static final String DEFAULT_SUMOLOGIC_URL = null;
+    private static final String DEFAULT_TRANSFORMER_CLASS = null;
+
+    public final String SUMOLOGIC_URL;
+    public final String TRANSFORMER_CLASS;
 
     /**
      * Configure the connector application with any set of properties that are unique to the application. Any
      * unspecified property will be set to a default value.
      */
     public KinesisConnectorForSumologicConfiguration(Properties properties, AWSCredentialsProvider credentialsProvider) {
         super(properties, credentialsProvider);
-        SUMOLOGIC_URL = properties.getProperty(PROP_SUMOLOGIC_URL, null);
-        SUMOLOGIC_USE_LOG4J = getBooleanProperty(PROP_SUMOLOGIC_USE_LOG4J, 
-            DEFAULT_SUMOLOGIC_USE_LOG4J, properties);
+        
+        SUMOLOGIC_URL = properties.getProperty(PROP_SUMOLOGIC_URL, DEFAULT_SUMOLOGIC_URL);
+        TRANSFORMER_CLASS = properties.getProperty(PROP_TRANSFORMER_CLASS, DEFAULT_TRANSFORMER_CLASS);
     }
-    
-    private boolean getBooleanProperty(String property, boolean defaultValue, Properties properties) {
-      String propertyValue = properties.getProperty(property, Boolean.toString(defaultValue));
-      return Boolean.parseBoolean(propertyValue);
-  }
 }

src/main/java/com/mcplusa/sumologic/SumologicMessageModelPipeline.java

@@ -1,9 +1,11 @@
 package com.mcplusa.sumologic;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import com.mcplusa.sumologic.KinesisMessageModel;
 import com.mcplusa.sumologic.KinesisMessageModelSumologicTransformer;
 import com.mcplusa.sumologic.implementations.SumologicEmitter;
-
 import com.amazonaws.services.kinesis.connectors.interfaces.IKinesisConnectorPipeline;
 import com.amazonaws.services.kinesis.connectors.KinesisConnectorConfiguration;
 import com.amazonaws.services.kinesis.connectors.impl.BasicMemoryBuffer;
@@ -27,6 +29,8 @@
 public class SumologicMessageModelPipeline implements
         IKinesisConnectorPipeline<KinesisMessageModel, String> {
 
+  private static final Log LOG = LogFactory.getLog(SumologicMessageModelPipeline.class);
+  
     @Override
     public IEmitter<String> getEmitter(KinesisConnectorConfiguration configuration) {
         return new SumologicEmitter(configuration);
@@ -40,7 +44,26 @@ public IBuffer<KinesisMessageModel> getBuffer(KinesisConnectorConfiguration conf
     @Override
     public ITransformer<KinesisMessageModel, String>
             getTransformer(KinesisConnectorConfiguration configuration) {
-        return new KinesisMessageModelSumologicTransformer();
+      
+      // Load specified class
+      String argClass = ((KinesisConnectorForSumologicConfiguration)configuration).TRANSFORMER_CLASS;
+      String className = "com.mcplusa.sumologic."+argClass;
+      ClassLoader classLoader = SumologicMessageModelPipeline.class.getClassLoader();
+      Class ModelClass = null;
+      try {
+        ModelClass = classLoader.loadClass(className);
+        ITransformer<KinesisMessageModel, String> ITransformerObject = (ITransformer<KinesisMessageModel, String>)ModelClass.newInstance();
+        LOG.info("Using transformer: "+ITransformerObject.getClass().getName());
+        return ITransformerObject;
+      } catch (ClassNotFoundException e) {
+        LOG.error("Class not found: "+className+" error: "+e.getMessage());
+      } catch (InstantiationException e) {
+        LOG.error("Class not found: "+className+" error: "+e.getMessage());
+      } catch (IllegalAccessException e) {
+        LOG.error("Class not found: "+className+" error: "+e.getMessage());
+      }
+      
+      return new DefaultKinesisMessageModelSumologicTransformer();
     }
 
     @Override

src/main/java/com/mcplusa/sumologic/SumologicSender.java

@@ -15,13 +15,10 @@
 import org.apache.http.util.EntityUtils;
 import org.apache.http.impl.client.DefaultHttpClient;
 import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
-import org.apache.log4j.Logger;
-import org.apache.log4j.PatternLayout;
-import org.apache.log4j.RollingFileAppender;
 
 public class SumologicSender {
   private static final Log LOG = LogFactory.getLog(SumologicSender.class);
- 
+
   private String url = null;  
   private HttpClient httpClient = null;
 
@@ -30,33 +27,23 @@ public class SumologicSender {
   private static final int RETRIES = 3;
   private static final int SLEEP_TIME = 1000;
 
-  private boolean useLog4j = false;
-  
-	public SumologicSender(String url, boolean useLog4j) {
+	    
+	public SumologicSender(String url) {
 		  this.url = url;
-		  this.useLog4j = useLog4j;
 
-		  if (!useLog4j) {
-    	    HttpParams params = new BasicHttpParams();
-    	    HttpConnectionParams.setConnectionTimeout(params, connectionTimeout);
-    	    HttpConnectionParams.setSoTimeout(params, socketTimeout);
-    	    httpClient = new DefaultHttpClient(new ThreadSafeClientConnManager(), params);
-		  }
-	}
-	
-	public boolean sendToSumologicUsingLog4j(String data) {
-	  Logger sumologicLog = Logger.getLogger("sumologic");
-    sumologicLog.trace(data);
-	  return true;
+	    HttpParams params = new BasicHttpParams();
+	    HttpConnectionParams.setConnectionTimeout(params, connectionTimeout);
+	    HttpConnectionParams.setSoTimeout(params, socketTimeout);
+	    httpClient = new DefaultHttpClient(new ThreadSafeClientConnManager(), params);
 	}
-	
-	public boolean sendToSumologicUsingHTTPRequest(String data) throws IOException {
+
+	public boolean sendToSumologic(String data) throws IOException{
 	  int retries = RETRIES;
-    int sleep_time = SLEEP_TIME;
-    int statusCode;
-    
-    do {
-      HttpPost post = null;
+	  int sleep_time = SLEEP_TIME;
+	  int statusCode;
+	  
+	  do {
+    	  HttpPost post = null;
       post = new HttpPost(url);
       post.setEntity(new StringEntity(data, HTTP.PLAIN_TEXT_TYPE, HTTP.UTF_8));
       HttpResponse response = httpClient.execute(post);
@@ -74,7 +61,7 @@ public boolean sendToSumologicUsingHTTPRequest(String data) throws IOException {
           Thread.sleep(sleep_time);
         } catch (InterruptedException ignore) {}
       }
-    } while (statusCode == 429 && retries > 0);
+	  } while (statusCode == 429 && retries > 0);
 
     // Check if the request was successful;
     if (statusCode != 200) {
@@ -85,13 +72,4 @@ public boolean sendToSumologicUsingHTTPRequest(String data) throws IOException {
       return true;
     } 
 	}
-
-	public boolean sendToSumologic(String data) throws IOException{
-	  if (this.useLog4j) {
-	    return sendToSumologicUsingLog4j(data);
-	  }
-	  else {
-	    return sendToSumologicUsingHTTPRequest(data);
-	  }
-	}
 }

src/main/java/com/mcplusa/sumologic/implementations/SumologicEmitter.java

@@ -23,29 +23,24 @@
  */
 public class SumologicEmitter implements IEmitter<String> {
     private static final Log LOG = LogFactory.getLog(SumologicEmitter.class);
-    
-    private static final boolean CONCATENATE_BATCH = false;
 
     private SumologicSender sender;
     private KinesisConnectorForSumologicConfiguration config;
 
     public SumologicEmitter(KinesisConnectorConfiguration configuration) {
         this.config = (KinesisConnectorForSumologicConfiguration) configuration;
-        sender = new SumologicSender(config.SUMOLOGIC_URL, config.SUMOLOGIC_USE_LOG4J);
+        sender = new SumologicSender(this.config.SUMOLOGIC_URL);
     }
 
-    public SumologicEmitter(String url, boolean useLog4j) {
-        sender = new SumologicSender(url, useLog4j);
+    public SumologicEmitter(String url) {
+        sender = new SumologicSender(url);
     }
 
     @Override
     public List<String> emit(final UnmodifiableBuffer<String> buffer)
         throws IOException {
         List<String> records = buffer.getRecords();
-        if (CONCATENATE_BATCH)
-          return sendBatchConcatenating(records);
-        else
-          return sendRecordsOneByOne(records);
+        return sendBatchConcatenating(records);
     }
 
     public List<String> sendBatchConcatenating(List<String> records) {
@@ -70,21 +65,6 @@ public List<String> sendBatchConcatenating(List<String> records) {
         return records;
       }
     }
-    
-    public List<String> sendRecordsOneByOne (List<String> records) {
-      ArrayList<String> failedRecords = new ArrayList<String>();
-      for (String record: records) {
-        try {
-          if (!sender.sendToSumologic(record)) {
-            failedRecords.add(record);
-          }
-        } catch (IOException e) {
-          LOG.warn("Couldn't send record: "+record);
-        }
-      }
-      LOG.info("Sent records: "+(records.size()-failedRecords.size())+" failed: "+failedRecords.size());
-      return failedRecords;
-    }
 
     @Override
     public void fail(List<String> records) {

src/test/java/com/mcplusa/sumologic/implementations/SumologicEmitterTest.java

@@ -39,7 +39,7 @@ public void theEmitterShouldReturnTheListParameterWhenFailing () {
     messages.add("This is message #3");
     messages.add("This is message #4");
 
-    SumologicEmitter emitter = new SumologicEmitter(url, false);
+    SumologicEmitter emitter = new SumologicEmitter(url);
     List <String> notEmittedMessages = emitter.sendBatchConcatenating(messages);
 
     Assert.assertEquals(messages, notEmittedMessages);
@@ -55,7 +55,7 @@ public void theEmitterShouldReturnAnEmptyListOnSuccess () {
     messages.add("This is message #3");
     messages.add("This is message #4");
 
-    SumologicEmitter emitter = new SumologicEmitter(url, false);
+    SumologicEmitter emitter = new SumologicEmitter(url);
     List <String> notEmittedMessages = emitter.sendBatchConcatenating(messages);
 
     Assert.assertEquals(0, notEmittedMessages.size());