SUMOK-17 Using Sumologics Log4j library

Author: Juan Pablo Diaz-Vaz

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/SumologicSender.java

@@ -15,9 +15,14 @@
 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 static final boolean USE_LOG4J = true;
 
   private String url = null;  
   private HttpClient httpClient = null;
@@ -27,7 +32,6 @@ public class SumologicSender {
   private static final int RETRIES = 3;
   private static final int SLEEP_TIME = 1000;
 
-	    
 	public SumologicSender(String url) {
 		  this.url = url;
 
@@ -36,8 +40,17 @@ public SumologicSender(String url) {
 	    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;
+	}
 
 	public boolean sendToSumologic(String data) throws IOException{
+	  if (USE_LOG4J)
+	    return sendToSumologicUsingLog4j(data);
+	  
 	  int retries = RETRIES;
 	  int sleep_time = SLEEP_TIME;
 	  int statusCode;

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

@@ -23,6 +23,8 @@
  */
 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;
@@ -40,7 +42,10 @@ public SumologicEmitter(String url) {
     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) {