Merged in SUMOK-17 (pull request #5)

Sumok 17

Author: michaelcizmar

log4j.properties

@@ -0,0 +1,16 @@
+# Root logger option
+log4j.rootLogger=INFO, stdout
+log4j.logger.sumologic = TRACE, sumo
+
+# Direct log messages to sumo
+log4j.appender.sumo=com.sumologic.log4j.BufferedSumoLogicAppender
+log4j.appender.sumo.url=https://collectors.us2.sumologic.com/receiver/v1/http/ZaVnC4dhaV0GzIY4tZaKLL26afV52gXBvSFc3jG1eLc2lKINzS2doZdRjUMQMb2CXK8r6fdmHoUazJiHjJ-2OygApoWFaxCTkWFrzAiraCc5i411pkio-g==
+log4j.appender.sumo.layout=org.apache.log4j.PatternLayout
+log4j.appender.sumo.layout.ConversionPattern=%d{DATE} %5p %c{1}:%L - %m%n
+log4j.additivity.sumo = false
+log4j.appender.sumo.Threshold = TRACE
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{DATE} %5p %c{1}:%L - %m%n
+log4j.appender.stdout.Threshold = INFO

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

@@ -14,15 +14,25 @@
 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 final boolean DEFAULT_SUMOLOGIC_USE_LOG4J = false;
 
     /**
      * 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);
     }
+    
+    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/SumologicSender.java

@@ -15,10 +15,13 @@
 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;
 
@@ -27,23 +30,33 @@ public class SumologicSender {
   private static final int RETRIES = 3;
   private static final int SLEEP_TIME = 1000;
 
-	    
-	public SumologicSender(String url) {
+  private boolean useLog4j = false;
+  
+	public SumologicSender(String url, boolean useLog4j) {
 		  this.url = url;
+		  this.useLog4j = useLog4j;
 
-	    HttpParams params = new BasicHttpParams();
-	    HttpConnectionParams.setConnectionTimeout(params, connectionTimeout);
-	    HttpConnectionParams.setSoTimeout(params, socketTimeout);
-	    httpClient = new DefaultHttpClient(new ThreadSafeClientConnManager(), params);
+		  if (!useLog4j) {
+    	    HttpParams params = new BasicHttpParams();
+    	    HttpConnectionParams.setConnectionTimeout(params, connectionTimeout);
+    	    HttpConnectionParams.setSoTimeout(params, socketTimeout);
+    	    httpClient = new DefaultHttpClient(new ThreadSafeClientConnManager(), params);
+		  }
 	}
-
-	public boolean sendToSumologic(String data) throws IOException{
+	
+	public boolean sendToSumologicUsingLog4j(String data) {
+	  Logger sumologicLog = Logger.getLogger("sumologic");
+    sumologicLog.trace(data);
+	  return true;
+	}
+	
+	public boolean sendToSumologicUsingHTTPRequest(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);
@@ -61,7 +74,7 @@ public boolean sendToSumologic(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) {
@@ -72,4 +85,13 @@ public boolean sendToSumologic(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,24 +23,29 @@
  */
 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(this.config.SUMOLOGIC_URL);
+        sender = new SumologicSender(config.SUMOLOGIC_URL, config.SUMOLOGIC_USE_LOG4J);
     }
 
-    public SumologicEmitter(String url) {
-        sender = new SumologicSender(url);
+    public SumologicEmitter(String url, boolean useLog4j) {
+        sender = new SumologicSender(url, useLog4j);
     }
 
     @Override
     public List<String> emit(final UnmodifiableBuffer<String> buffer)
         throws IOException {
         List<String> records = buffer.getRecords();
-        return sendBatchConcatenating(records);
+        if (CONCATENATE_BATCH)
+          return sendBatchConcatenating(records);
+        else
+          return sendRecordsOneByOne(records);
     }
 
     public List<String> sendBatchConcatenating(List<String> records) {
@@ -65,6 +70,21 @@ 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);
+    SumologicEmitter emitter = new SumologicEmitter(url, false);
     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);
+    SumologicEmitter emitter = new SumologicEmitter(url, false);
     List <String> notEmittedMessages = emitter.sendBatchConcatenating(messages);
 
     Assert.assertEquals(0, notEmittedMessages.size());