Release of the CloudWatch Logs Subscription Consumer, with samples th…

…at connect to Stdout, Elasticsearch and Amazon S3

.gitignore

@@ -0,0 +1,2 @@
+target/
+AwsCredentials.properties

LICENSE.txt

@@ -0,0 +1,39 @@
+
+Amazon Software License
+
+This Amazon Software License (“License”) governs your use, reproduction, and distribution of the accompanying software as specified below.
+1. Definitions
+
+“Licensor” means any person or entity that distributes its Work.
+
+“Software” means the original work of authorship made available under this License.
+
+“Work” means the Software and any additions to or derivative works of the Software that are made available under this License.
+
+The terms “reproduce,” “reproduction,” “derivative works,” and “distribution” have the meaning as provided under U.S. copyright law; provided, however, that for the purposes of this License, derivative works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work.
+
+Works, including the Software, are “made available” under this License by including in or with the Work either (a) a copyright notice referencing the applicability of this License to the Work, or (b) a copy of this License.
+2. License Grants
+
+2.1 Copyright Grant. Subject to the terms and conditions of this License, each Licensor grants to you a perpetual, worldwide, non-exclusive, royalty-free, copyright license to reproduce, prepare derivative works of, publicly display, publicly perform, sublicense and distribute its Work and any resulting derivative works in any form.
+
+2.2 Patent Grant. Subject to the terms and conditions of this License, each Licensor grants to you a perpetual, worldwide, non-exclusive, royalty-free patent license to make, have made, use, sell, offer for sale, import, and otherwise transfer its Work, in whole or in part. The foregoing license applies only to the patent claims licensable by Licensor that would be infringed by Licensor’s Work (or portion thereof) individually and excluding any combinations with any other materials or technology.
+3. Limitations
+
+3.1 Redistribution. You may reproduce or distribute the Work only if (a) you do so under this License, (b) you include a complete copy of this License with your distribution, and (c) you retain without modification any copyright, patent, trademark, or attribution notices that are present in the Work.
+
+3.2 Derivative Works. You may specify that additional or different terms apply to the use, reproduction, and distribution of your derivative works of the Work (“Your Terms”) only if (a) Your Terms provide that the use limitation in Section 3.3 applies to your derivative works, and (b) you identify the specific derivative works that are subject to Your Terms. Notwithstanding Your Terms, this License (including the redistribution requirements in Section 3.1) will continue to apply to the Work itself.
+
+3.3 Use Limitation. The Work and any derivative works thereof only may be used or intended for use with the web services, computing platforms or applications provided by Amazon.com, Inc. or its affiliates, including Amazon Web Services, Inc.
+
+3.4 Patent Claims. If you bring or threaten to bring a patent claim against any Licensor (including any claim, cross-claim or counterclaim in a lawsuit) to enforce any patents that you allege are infringed by any Work, then your rights under this License from such Licensor (including the grants in Sections 2.1 and 2.2) will terminate immediately.
+
+3.5 Trademarks. This License does not grant any rights to use any Licensor’s or its affiliates’ names, logos, or trademarks, except as necessary to reproduce the notices described in this License.
+
+3.6 Termination. If you violate any term of this License, then your rights under this License (including the grants in Sections 2.1 and 2.2) will terminate immediately.
+4. Disclaimer of Warranty.
+
+THE WORK IS PROVIDED “AS IS” WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WARRANTIES OR CONDITIONS OF M ERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, TITLE OR NON-INFRINGEMENT. YOU BEAR THE RISK OF UNDERTAKING ANY ACTIVITIES UNDER THIS LICENSE. SOME STATES’ CONSUMER LAWS DO NOT ALLOW EXCLUSION OF AN IMPLIED WARRANTY, SO THIS DISCLAIMER MAY NOT APPLY TO YOU.
+5. Limitation of Liability.
+
+EXCEPT AS PROHIBITED BY APPLICABLE LAW, IN NO EVENT AND UNDER NO LEGAL THEORY, WHETHER IN TORT (INCLUDING NEGLIGENCE), CONTRACT, OR OTHERWISE SHALL ANY LICENSOR BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF OR RELATED TO THIS LICENSE, THE USE OR INABILITY TO USE THE WORK (INCLUDING BUT NOT LIMITED TO LOSS OF GOODWILL, BUSINESS INTERRUPTION, LOST PROFITS OR DATA, COMPUTER FAILURE OR MALFUNCTION, OR ANY OTHER COMM ERCIAL DAMAGES OR LOSSES), EVEN IF THE LICENSOR HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES.

NOTICE.txt

@@ -0,0 +1,2 @@
+CloudWatch Logs Subscription Consumer
+Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.

README.md

@@ -0,0 +1,2 @@
+# cloudwatch-logs-subscription-consumer
+A Java application that consumes log data emitted to Amazon Kinesis from a CloudWatch Logs Subscription Filter: http://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/Subscriptions.html

pom.xml

@@ -0,0 +1,133 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.amazonaws</groupId>
+    <artifactId>cloudwatch-logs-subscription-consumer</artifactId>
+    <version>1.0.0</version>
+    <name>CloudWatch Logs Subscription Consumer</name>
+    <description>The CloudWatch Logs Subscription Consumer helps Java developers consume a real-time feed of CloudWatch Logs data for custom processing, analysis, or loading to other systems.</description>
+    <url>https://aws.amazon.com/cloudwatch</url>
+
+    <licenses>
+        <license>
+           <name>Amazon Software License</name>
+           <url>https://aws.amazon.com/asl</url>
+           <distribution>repo</distribution>
+        </license>
+    </licenses>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>amazon-kinesis-connectors</artifactId>
+            <version>LATEST</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>1.6.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>LATEST</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <developers>
+        <developer>
+            <id>amazonwebservices</id>
+            <organization>Amazon Web Services</organization>
+            <organizationUrl>https://aws.amazon.com</organizationUrl>
+            <roles>
+                <role>developer</role>
+            </roles>
+        </developer>
+    </developers>
+
+    <build>
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-compiler-plugin</artifactId>
+                    <version>3.2</version>
+                    <configuration>
+                        <source>1.7</source>
+                        <target>1.7</target>
+                        <encoding>UTF-8</encoding>
+                    </configuration>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-gpg-plugin</artifactId>
+                <version>1.5</version>
+                <executions>
+                    <execution>
+                        <id>sign-artifacts</id>
+                        <phase>verify</phase>
+                        <goals>
+                            <goal>sign</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>exec-maven-plugin</artifactId>
+                <version>1.2.1</version>
+                <configuration>
+                    <mainClass>com.amazonaws.services.logs.connectors.samples.elasticsearch.ElasticsearchConnector</mainClass>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <version>2.10</version>
+                <executions>
+                    <execution>
+                        <id>copy-dependencies</id>
+                        <phase>prepare-package</phase>
+                        <goals>
+                            <goal>copy-dependencies</goal>
+                        </goals>
+                        <configuration>
+                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
+                            <overWriteReleases>false</overWriteReleases>
+                            <overWriteSnapshots>false</overWriteSnapshots>
+                            <overWriteIfNewer>true</overWriteIfNewer>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>2.6</version>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <classpathPrefix>lib/</classpathPrefix>
+                            <mainClass>com.amazonaws.services.logs.connectors.samples.elasticsearch.ElasticsearchConnector</mainClass>
+                        </manifest>
+                    </archive>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

src/main/java/com/amazonaws/services/logs/connectors/elasticsearch/CloudWatchLogsElasticsearchDocument.java

@@ -0,0 +1,188 @@
+/*
+ * Copyright 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.amazonaws.services.logs.connectors.elasticsearch;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.elasticsearch.common.lang3.StringUtils;
+
+import com.amazonaws.services.logs.subscriptions.CloudWatchLogsEvent;
+import com.amazonaws.util.json.JSONException;
+import com.amazonaws.util.json.JSONObject;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonRawValue;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * Defines the log event structure that would be sent to Elasticsearch.
+ */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class CloudWatchLogsElasticsearchDocument implements Serializable {
+
+    private static final long serialVersionUID = -8898041979675208782L;
+
+    private static final ObjectMapper JSON_OBJECT_MAPPER = new ObjectMapper();
+
+    private static final Log LOG = LogFactory.getLog(CloudWatchLogsElasticsearchDocument.class);
+
+    private String id;
+    private long timestamp;
+    private String message;
+    private String fields;
+
+    private String owner;
+    private String logGroup;
+    private String logStream;
+
+    public CloudWatchLogsElasticsearchDocument() {
+        // required by the Jackson JSON mapper
+    }
+
+    public CloudWatchLogsElasticsearchDocument(CloudWatchLogsEvent event) {
+        this.id = event.getId();
+        this.timestamp = event.getTimestamp();
+        this.message = event.getMessage();
+        this.fields = getFields(event.getMessage(), event.getExtractedFields());
+        this.owner = event.getOwner();
+        this.logGroup = event.getLogGroup();
+        this.logStream = event.getLogStream();
+    }
+
+    /**
+     *
+     * Determines which additional fields get put into the Elasticsearch document.
+     */
+    private String getFields(String message, Map<String, String> extractedFields) {
+        // if extractedFields are available from CloudWatch Logs, use them as Elasticsearch fields
+        if (extractedFields != null && extractedFields.size() > 0) {
+            JSONObject extractedFieldsInJson = new JSONObject();
+
+            try {
+                for (Map.Entry<String, String> entry : extractedFields.entrySet()) {
+                    String fieldName = entry.getKey();
+                    String value = entry.getValue();
+
+                    if (StringUtils.isNumeric(value)) {
+                        extractedFieldsInJson.put(fieldName, Double.parseDouble(value));
+                    } else {
+                        extractedFieldsInJson.put(fieldName, value);
+                    }
+                }
+
+                return extractedFieldsInJson.toString();
+            } catch (JSONException e) {
+                LOG.error("Unable to convert extractedFields to JSON: " + e.getMessage());
+                return null;
+            }
+        }
+
+        // if the message is valid JSON, use the message as is for Elasticsearch fields
+        if (isMessageValidJson(message)) {
+            return message;
+        }
+
+        // if there are no extractedFields and the message is not valid JSON, don't emit any Elasticsearch fields
+        return null;
+    }
+
+    public static boolean isMessageValidJson(String message) {
+
+        try {
+            JSON_OBJECT_MAPPER.readTree(message);
+        } catch (IOException e) {
+            return false;
+        }
+        return true;
+    }
+
+    @JsonProperty("@id")
+    public String getId() {
+        return id;
+    }
+
+    @JsonProperty("@timestamp")
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    @JsonProperty("@message")
+    public String getMessage() {
+        return message;
+    }
+
+    @JsonRawValue
+    @JsonProperty("$")
+    public String getFields() {
+        return fields;
+    }
+
+    @JsonProperty("@owner")
+    public String getOwner() {
+        return owner;
+    }
+
+    @JsonProperty("@log_group")
+    public String getLogGroup() {
+        return logGroup;
+    }
+
+    @JsonProperty("@log_stream")
+    public String getLogStream() {
+        return logStream;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public void setTimestamp(long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    public void setFields(String fields) {
+        this.fields = fields;
+    }
+
+    public void setOwner(String owner) {
+        this.owner = owner;
+    }
+
+    public void setLogGroup(String logGroup) {
+        this.logGroup = logGroup;
+    }
+
+    public void setLogStream(String logStream) {
+        this.logStream = logStream;
+    }
+
+    @Override
+    public String toString() {
+        try {
+            return new ObjectMapper().writeValueAsString(this);
+        } catch (JsonProcessingException e) {
+            return super.toString();
+        }
+    }
+}