SUMOK-15 Concatenating records to send in batch

Author: Juan Pablo Diaz-Vaz

src/com/mcplusa/kinesis/implementations/SumologicTransformer.java

@@ -1,17 +1,3 @@
-/*
- * Copyright 2013-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Licensed under the Amazon Software License (the "License").
- * You may not use this file except in compliance with the License.
- * A copy of the License is located at
- *
- * http://aws.amazon.com/asl/
- *
- * or in the "license" file accompanying this file. This file is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
- * express or implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
 package com.mcplusa.sumologic;
 
 import java.util.Properties;

src/com/mcplusa/sumologic/KinesisConnectorForSumologicConfiguration.java

@@ -1,18 +1,4 @@
-/*
- * Copyright 2013-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Licensed under the Amazon Software License (the "License").
- * You may not use this file except in compliance with the License.
- * A copy of the License is located at
- *
- * http://aws.amazon.com/asl/
- *
- * or in the "license" file accompanying this file. This file is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
- * express or implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
-package com.mcplusa.kinesis;
+package com.mcplusa.sumologic;
 
 import java.io.Serializable;
 

…mcplusa/kinesis/KinesisMessageModel.java …plusa/sumologic/KinesisMessageModel.javasrc/com/mcplusa/kinesis/KinesisMessageModel.java renamed to src/com/mcplusa/sumologic/KinesisMessageModel.java src/com/mcplusa/kinesis/KinesisMessageModel.java renamed to src/com/mcplusa/sumologic/KinesisMessageModel.java

@@ -1,8 +1,9 @@
 package com.mcplusa.sumologic;
 
 import com.amazonaws.services.kinesis.connectors.BasicJsonTransformer;
-import com.mcplusa.kinesis.KinesisMessageModel;
-import com.mcplusa.kinesis.implementations.SumologicTransformer;
+
+import com.mcplusa.sumologic.KinesisMessageModel;
+import com.mcplusa.sumologic.implementations.SumologicTransformer;
 
 
 /**

src/com/mcplusa/sumologic/KinesisMessageModelSumologicTransformer.java

@@ -3,7 +3,7 @@
 import com.amazonaws.services.kinesis.connectors.KinesisConnectorRecordProcessorFactory;
 
 import com.mcplusa.kinesis.KinesisConnectorExecutor;
-import com.mcplusa.kinesis.KinesisMessageModel;
+import com.mcplusa.sumologic.KinesisMessageModel;
 import com.mcplusa.sumologic.SumologicMessageModelPipeline;
 
 public class SumologicExecutor extends KinesisConnectorExecutor<KinesisMessageModel, String> {

src/com/mcplusa/sumologic/SumologicExecutor.java

@@ -1,8 +1,8 @@
 package com.mcplusa.sumologic;
 
-import com.mcplusa.kinesis.KinesisMessageModel;
+import com.mcplusa.sumologic.KinesisMessageModel;
 import com.mcplusa.sumologic.KinesisMessageModelSumologicTransformer;
-import com.mcplusa.kinesis.implementations.SumologicEmitter;
+import com.mcplusa.sumologic.implementations.SumologicEmitter;
 
 import com.amazonaws.services.kinesis.connectors.interfaces.IKinesisConnectorPipeline;
 import com.amazonaws.services.kinesis.connectors.KinesisConnectorConfiguration;

src/com/mcplusa/sumologic/SumologicMessageModelPipeline.java

@@ -36,21 +36,20 @@ public SumologicSender(String url) {
 	    httpClient = new DefaultHttpClient(new ThreadSafeClientConnManager(), params);
 	}
 
-	public void sendToSumologic(String data) {
+	public void sendToSumologic(String data) throws IOException{
   	  HttpPost post = null;
-    try {
-      post = new HttpPost(url);
-      post.setEntity(new StringEntity(data, HTTP.PLAIN_TEXT_TYPE, HTTP.UTF_8));
-      HttpResponse response = httpClient.execute(post);
-      int statusCode = response.getStatusLine().getStatusCode();
-      if (statusCode != 200) {
-        LOG.warn(String.format("Received HTTP error from Sumo Service: %d", statusCode));
-      }
-      //need to consume the body if you want to re-use the connection.
-      EntityUtils.consume(response.getEntity());
-    } catch (IOException e) {
-      LOG.warn("Could not send log to Sumo Logic", e);
-      try { post.abort(); } catch (Exception ignore) {}
+    post = new HttpPost(url);
+    post.setEntity(new StringEntity(data, HTTP.PLAIN_TEXT_TYPE, HTTP.UTF_8));
+    HttpResponse response = httpClient.execute(post);
+    int statusCode = response.getStatusLine().getStatusCode();
+    if (statusCode != 200) {
+      LOG.warn(String.format("Received HTTP error from Sumo Service: %d", statusCode));
     }
+    
+    //need to consume the body if you want to re-use the connection.
+    EntityUtils.consume(response.getEntity());
+    try { 
+      post.abort(); 
+    } catch (Exception ignore) {}
     }
 }

src/com/mcplusa/sumologic/SumologicSender.java

@@ -1,18 +1,4 @@
-/**
- * Copyright 2013-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Licensed under the Amazon Software License (the "License").
- * You may not use this file except in compliance with the License.
- * A copy of the License is located at
- *
- * http://aws.amazon.com/asl/
- *
- * or in the "license" file accompanying this file. This file is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
- * express or implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
-package com.mcplusa.kinesis.implementations;
+package com.mcplusa.sumologic.implementations;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -51,12 +37,24 @@ public List<String> emit(final UnmodifiableBuffer<String> buffer)
         throws IOException {
 
         List<String> records = buffer.getRecords();
-        for (String record : records) {
-            LOG.info("Got record: "+record);
-            sender.sendToSumologic(record); // TODO send in batches
-        }
-        
-        return new ArrayList<String>(); // TODO return unprocessed records
+
+        return sendBatchConcatenating(records);
+    }
+    
+    public List<String> sendBatchConcatenating(List<String> records) {
+      String message = "";
+      for(String record: records) {
+        message += record;
+        message += "\n";
+      }
+      try {
+        LOG.info("Sending batch of: "+records.size()+" records");
+        sender.sendToSumologic(message);
+      } catch (IOException e) {
+        LOG.warn("Couldn't send record to Sumologic: "+e.getMessage());
+        return records;
+      }
+      return new ArrayList<String>();
     }
 
     @Override

…is/implementations/SumologicEmitter.java …ic/implementations/SumologicEmitter.javasrc/com/mcplusa/kinesis/implementations/SumologicEmitter.java renamed to src/com/mcplusa/sumologic/implementations/SumologicEmitter.java src/com/mcplusa/kinesis/implementations/SumologicEmitter.java renamed to src/com/mcplusa/sumologic/implementations/SumologicEmitter.java

@@ -0,0 +1,13 @@
+package com.mcplusa.sumologic.implementations;
+
+import com.amazonaws.services.kinesis.connectors.interfaces.ITransformer;
+
+/**
+ * This interface defines an ITransformer that has an output type of Map of attribute name (String)
+ * to AttributeValue so that the item can be put into Sumologic.
+ * 
+ * @param <T>
+ */
+public interface SumologicTransformer<T> extends ITransformer<T, String> {
+
+}