initial version

.gitignore

@@ -0,0 +1,15 @@
+/.bundle/
+/.yardoc
+/Gemfile.lock
+/_yardoc/
+/coverage/
+/doc/
+/pkg/
+/spec/reports/
+/tmp/
+*.bundle
+*.so
+*.o
+*.a
+mkmf.log
+lib/logstash-input-kinesis_jars.rb

Gemfile

@@ -0,0 +1,4 @@
+source 'https://rubygems.org'
+
+# Specify your gem's dependencies in logstash-input-kinesis.gemspec
+gemspec

LICENSE.txt

@@ -0,0 +1,22 @@
+Copyright (c) 2015 Brian Palmer
+
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,31 @@
+# Logstash::Input::Kinesis
+
+TODO: Write a gem description
+
+## Installation
+
+Add this line to your application's Gemfile:
+
+```ruby
+gem 'logstash-input-kinesis'
+```
+
+And then execute:
+
+    $ bundle
+
+Or install it yourself as:
+
+    $ gem install logstash-input-kinesis
+
+## Usage
+
+TODO: Write usage instructions here
+
+## Contributing
+
+1. Fork it ( https://github.com/[my-github-username]/logstash-input-kinesis/fork )
+2. Create your feature branch (`git checkout -b my-new-feature`)
+3. Commit your changes (`git commit -am 'Add some feature'`)
+4. Push to the branch (`git push origin my-new-feature`)
+5. Create a new Pull Request

Rakefile

@@ -0,0 +1,2 @@
+require "bundler/gem_tasks"
+

lib/logstash/inputs/kinesis.rb

@@ -0,0 +1,90 @@
+# encoding: utf-8
+require "logstash/inputs/base"
+require "logstash/errors"
+require "logstash/environment"
+require "logstash/namespace"
+
+require "logstash/inputs/kinesis/version"
+require 'logstash-input-kinesis_jars'
+
+class LogStash::Inputs::Kinesis < LogStash::Inputs::Base
+  KCL = com.amazonaws.services.kinesis.clientlibrary.lib.worker
+
+  config_name 'kinesis'
+  milestone 1
+
+  # The application name used for the dynamodb coordination table. Must be
+  # unique for this kinesis stream.
+  config :application_name, :validate => :string, :default => "logstash"
+
+  # The kinesis stream name.
+  config :kinesis_stream_name, :validate => :string, :required => true
+
+  def register
+    # the INFO log level is extremely noisy in KCL
+    org.apache.commons.logging::LogFactory.getLog("com.amazonaws.services.kinesis").
+      logger.setLevel(java.util.logging::Level::WARNING)
+
+    worker_id = java.util::UUID.randomUUID.to_s
+    creds = com.amazonaws.auth::DefaultAWSCredentialsProviderChain.new()
+    @config = KCL::KinesisClientLibConfiguration.new(
+      @application_name,
+      @kinesis_stream_name,
+      creds,
+      worker_id).withInitialPositionInStream(KCL::InitialPositionInStream::TRIM_HORIZON)
+  end
+
+  def run(output_queue)
+    worker_factory = WorkerFactory.new(@codec, output_queue, method(:decorate))
+    @worker = KCL::Worker.new(worker_factory, @config)
+    @worker.run()
+  end
+
+  def teardown
+    @worker.shutdown if @worker
+  end
+
+  class WorkerFactory
+    include com.amazonaws.services.kinesis.clientlibrary.interfaces::IRecordProcessorFactory
+    def initialize(codec, output_queue, decorator)
+      @codec = codec
+      @output_queue = output_queue
+      @decorator = decorator
+    end
+
+    def createProcessor
+      Worker.new(@codec.clone, @output_queue, @decorator)
+    end
+  end
+
+  class Worker
+    include com.amazonaws.services.kinesis.clientlibrary.interfaces::IRecordProcessor
+
+    def initialize(*args)
+      # nasty hack, because this is the name of a method on IRecordProcessor, but also ruby's constructor
+      if !@constructed
+        @codec, @output_queue, @decorator = args
+        @constructed = true
+      else
+        _shard_id, _ = args
+        @decoder = java.nio.charset::Charset.forName("UTF-8").newDecoder()
+      end
+    end
+
+    def processRecords(records, checkpointer)
+      records.each { |record| process_record(record) }
+      checkpointer.checkpoint()
+    end
+
+    def shutdown(checkpointer, reason)
+    end
+
+    def process_record(record)
+      raw = @decoder.decode(record.getData).to_s
+      @codec.decode(raw) do |event|
+        @decorator.call(event)
+        @output_queue << event
+      end
+    end
+  end
+end

lib/logstash/inputs/kinesis/version.rb

@@ -0,0 +1,7 @@
+module Logstash
+  module Input
+    module Kinesis
+      VERSION = "1.0.0"
+    end
+  end
+end

logstash-input-kinesis.gemspec

@@ -0,0 +1,33 @@
+# coding: utf-8
+lib = File.expand_path('../lib', __FILE__)
+$LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
+require 'logstash/inputs/kinesis/version'
+
+Gem::Specification.new do |spec|
+  spec.name          = "logstash-input-kinesis"
+  spec.version       = Logstash::Input::Kinesis::VERSION
+  spec.authors       = ["Brian Palmer"]
+  spec.email         = ["[email protected]"]
+  spec.summary       = %q{Logstash plugin for Kinesis input}
+  spec.description   = %q{This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program}
+  spec.homepage      = ""
+  spec.license       = "MIT"
+
+  spec.files         = %w[Gemfile LICENSE.txt README.md Rakefile] + Dir.glob("lib/logstash/inputs/**/*")
+  spec.test_files    = spec.files.grep(%r{^(test|spec|features)/})
+  spec.require_paths = ["lib"]
+
+  # Special flag to let us know this is actually a logstash plugin
+  spec.metadata      = { "logstash_plugin" => "true", "logstash_group" => "input" }
+
+  spec.platform      = 'java'
+
+  spec.requirements << "jar 'com.amazonaws:amazon-kinesis-client', '1.2.1'"
+
+  spec.add_runtime_dependency 'jar-dependencies', '0.1.7'
+  spec.add_runtime_dependency 'ruby-maven', '3.1.1.0.8'
+  spec.add_runtime_dependency "maven-tools", '1.0.7'
+
+  spec.add_development_dependency "bundler", "~> 1.7"
+  spec.add_development_dependency "rake", "~> 10.0"
+end