New kafka

README.md

@@ -47,7 +47,9 @@ file_logger = LogStashLogger.new(type: :file, path: 'log/development.log', sync:
 unix_logger = LogStashLogger.new(type: :unix, path: '/tmp/sock')
 syslog_logger = LogStashLogger.new(type: :syslog)
 redis_logger = LogStashLogger.new(type: :redis)
-kafka_logger = LogStashLogger.new(type: :kafka)
+** NOTE: The current kafka class will be deprecated in a future version.
+  For now, migrate to using kafka_new**
+kafka_logger = LogStashLogger.new(type: :kafka_new)
 stdout_logger = LogStashLogger.new(type: :stdout)
 stderr_logger = LogStashLogger.new(type: :stderr)
 io_logger = LogStashLogger.new(type: :io, io: io)
@@ -454,26 +456,37 @@ config.logstash.port = 6379
 
 #### Kafka
 
-Add the poseidon gem to your Gemfile:
+Add the ruby-kafka gem to your Gemfile:
 
-    gem 'poseidon'
+    gem 'ruby-kafka'
 
 ```ruby
+## NOTE: A future version of this gem will remove the current
+# implementation of the kafka client. This will be a breaking change. Use
+# kafka_new to ensure forward compatibility
 # Required
-config.logstash.type = :kafka
 
-# Optional, will default to the 'logstash' topic
-config.logstash.path = 'logstash'
-
-# Optional, will default to the 'logstash-logger' producer
-config.logstash.producer = 'logstash-logger'
-
-# Optional, will default to localhost:9092 host/port
-config.logstash.hosts = ['localhost:9092']
-
-# Optional, will default to 1s backoff
-config.logstash.backoff = 1
+config.logstash.type = :kafka_new
 
+# Required
+config.logstash.topic = 'logstash-topic'
+
+# Required, can be in one of two formats:
+# String format (splits on single space):
+config.logstash.brokers = 'localhost:9092 some-other-host.net:9300'
+# Array format
+config.logstash.brokers = %w(localhost:9092 some-other-host.net:9300)
+
+# Optional, defaults to 'ruby-kafka'
+config.logstash.client_id = 'logstash-client-alpha'
+
+# Optional, transmit over TLS
+# NOTE: either 0 or all 3 ssl_parameters must be provided for a
+# successful connection. An exception will be raised if 1 or 2 params
+# are povided
+config.logstash.ssl_ca_cert: ENV['CLOUDKAFKA_CA']
+config.logstash.ssl_client_cert: ENV['CLOUDKAFKA_CERT']
+config.logstash.ssl_client_cert_key: ENV['CLOUDKAFKA_PRIVATE_KEY']
 ```
 
 #### File

lib/logstash-logger/device.rb

@@ -12,6 +12,7 @@ module Device
     autoload :Unix, 'logstash-logger/device/unix'
     autoload :Redis, 'logstash-logger/device/redis'
     autoload :Kafka, 'logstash-logger/device/kafka'
+    autoload :KafkaNew, 'logstash-logger/device/kafka_new'
     autoload :File, 'logstash-logger/device/file'
     autoload :IO, 'logstash-logger/device/io'
     autoload :Stdout, 'logstash-logger/device/stdout'
@@ -50,6 +51,7 @@ def self.device_klass_for(type)
         when :file then File
         when :redis then Redis
         when :kafka then Kafka
+        when :kafka_new then KafkaNew
         when :io then IO
         when :stdout then Stdout
         when :stderr then Stderr

lib/logstash-logger/device/kafka.rb

@@ -12,7 +12,15 @@ class Kafka < Connectable
 
       attr_accessor :hosts, :topic, :producer, :backoff
 
+      @@deprecation_message = <<-MSG
+          [DEPRECATION WARNING]
+          Poseidon client will be deprecated and requires different configuration
+          parameters (but they are similar). Update your Kafka configuration to
+          use :kafka_new to ensure forward compatibility
+        MSG
+
       def initialize(opts)
+        warn @@deprecation_message
         super
         host = opts[:host] || DEFAULT_HOST
         port = opts[:port] || DEFAULT_PORT

lib/logstash-logger/device/kafka_new.rb

@@ -0,0 +1,115 @@
+module LogStashLogger
+  module Device
+    class KafkaNew < Connectable
+      class TLSConfiguration
+        attr_reader :ssl_ca_cert, :ssl_client_cert, :ssl_client_cert_key
+
+        def initialize(opts = {})
+          @ssl_ca_cert = opts[:ssl_ca_cert]
+          @ssl_client_cert = opts[:ssl_client_cert]
+          @ssl_client_cert_key = opts[:ssl_client_cert_key]
+        end
+
+        def cert_bundle
+          @cert_bundle ||= all_cert_params? ? cert_params_as_hash : {}
+        end
+
+        def valid?
+          all_cert_params? || no_cert_params?
+        end
+
+        def invalid?
+          !valid?
+        end
+
+        private
+
+        def cert_params_as_hash
+          { ssl_ca_cert: @ssl_ca_cert,
+            ssl_client_cert: @ssl_client_cert,
+            ssl_client_cert_key: @ssl_client_cert_key,
+          }
+        end
+
+        def all_cert_params?
+          cert_params_as_hash.values.compact.length == valid_cert_params_length
+        end
+
+        def no_cert_params?
+          cert_params_as_hash.values.compact.empty?
+        end
+
+        def valid_cert_params_length
+          cert_params_as_hash.keys.length
+        end
+      end
+
+      attr_reader :topic, :brokers, :cert_bundle, :kafka_tls_configurator,
+        :client_id
+
+      def initialize(opts = {}, kafka_tls_configurator = TLSConfiguration)
+        require 'ruby-kafka'
+        super(opts)
+
+        @client_id = opts[:client_id]
+        @topic = opts[:topic] || raise_no_topic_set!
+        @kafka_tls_configurator = kafka_tls_configurator
+        @brokers = make_brokers_array(opts[:brokers])
+        make_cert_bundle(opts)
+      end
+
+      def connection
+        @connection ||= ::Kafka.new(kafka_client_connection_hash)
+      end
+
+      def write_one(message, topic=nil)
+        topic ||= @topic
+        write_messages_to_broker_and_deliver do |producer|
+          producer.produce(message, topic: topic)
+        end
+      end
+
+      def write_batch(messages, topic=nil)
+        topic ||= @topic
+        write_messages_to_broker_and_deliver do |producer|
+          messages.each {|msg| producer.produce(msg, topic: topic) }
+        end
+      end
+
+      private
+
+      def write_messages_to_broker_and_deliver(&block)
+        kproducer = connection.producer
+        block.call(kproducer) if block_given?
+        kproducer.deliver_messages
+      end
+
+      def kafka_client_connection_hash
+        { seed_brokers: @brokers,
+          client_id: @client_id,
+        }.merge(@cert_bundle)
+      end
+
+      def raise_no_topic_set!
+        fail ArgumentError, "a topic must be configured"
+      end
+
+      def make_brokers_array(opt)
+        case opt
+        when Array
+          opt
+        when String
+          opt.split("\s")
+        end
+      end
+
+      def make_cert_bundle(opts)
+        tls_conf = kafka_tls_configurator.new(opts)
+        if tls_conf.invalid?
+          fail ArgumentError, "all ssl parameters (ssl_ca_cert, ssl_client_cert and ssl_client_cert_key) are required or do use any of them to not use TLS"
+        end
+        @cert_bundle ||= tls_conf.cert_bundle
+      end
+    end
+  end
+end

lib/logstash-logger/version.rb

@@ -1,3 +1,3 @@
 module LogStashLogger
-  VERSION = "0.19.2"
+  VERSION = "0.19.3"
 end

logstash-logger.gemspec

@@ -26,6 +26,7 @@ Gem::Specification.new do |gem|
   end
   gem.add_development_dependency 'redis'
   gem.add_development_dependency 'poseidon'
+  gem.add_development_dependency 'ruby-kafka'
 
   if RUBY_VERSION < '2' || defined?(JRUBY_VERSION)
     gem.add_development_dependency 'SyslogLogger'