Module: Fluent::KafkaPluginUtil::SSLSettings

Included in:
Fluent::Kafka2Output, Fluent::KafkaGroupInput, Fluent::KafkaInput, Fluent::KafkaOutput, Fluent::KafkaOutputBuffered, Fluent::KafkaOutputBuffered2, Plugin::RdKafkaGroupInput, Rdkafka2Output
Defined in:
lib/fluent/plugin/kafka_plugin_util.rb

Constant Summary collapse

DummyFormatter = 
Object.new

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(klass) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/fluent/plugin/kafka_plugin_util.rb', line 17

def self.included(klass)
  klass.instance_eval {
    # https://github.com/zendesk/ruby-kafka#encryption-and-authentication-using-ssl
    config_param :ssl_ca_cert, :array, :value_type => :string, :default => nil,
                 :desc => "a PEM encoded CA cert to use with and SSL connection."
    config_param :ssl_client_cert, :string, :default => nil,
                 :desc => "a PEM encoded client cert to use with and SSL connection. Must be used in combination with ssl_client_cert_key."
    config_param :ssl_client_cert_key, :string, :default => nil,
                 :desc => "a PEM encoded client cert key to use with and SSL connection. Must be used in combination with ssl_client_cert."
    config_param :ssl_client_cert_key_password, :string, :default => nil, secret: true,
                 :desc => "a PEM encoded client cert key password to use with SSL connection."
    config_param :ssl_client_cert_chain, :string, :default => nil,
                 :desc => "an extra PEM encoded cert to use with and SSL connection."
    config_param :ssl_ca_certs_from_system, :bool, :default => false,
                 :desc => "this configures the store to look up CA certificates from the system default certificate store on an as needed basis. The location of the store can usually be determined by: OpenSSL::X509::DEFAULT_CERT_FILE."
    config_param :ssl_verify_hostname, :bool, :default => true,
                 :desc => "this configures whether hostname of certificate should be verified or not."
  }
end

Instance Method Details

#pickup_ssl_endpoint(node) ⇒ Object 



60
61
62
63
64
 
# File 'lib/fluent/plugin/kafka_plugin_util.rb', line 60

def pickup_ssl_endpoint(node)
  ssl_endpoint = node['endpoints'].find {|e| e.start_with?('SSL')}
  raise 'no SSL endpoint found on Zookeeper' unless ssl_endpoint
  return [URI.parse(ssl_endpoint).host, URI.parse(ssl_endpoint).port].join(':')
end

#read_ssl_file(path) ⇒ Object 



50
51
52
53
54
55
56
57
58
 
# File 'lib/fluent/plugin/kafka_plugin_util.rb', line 50

def read_ssl_file(path)
  return nil if path.nil? || path.respond_to?(:strip) && path.strip.empty?

  if path.is_a?(Array)
    path.map { |fp| File.read(fp) }
  else
    File.read(path)
  end
end

#start ⇒ Object 



39
40
41
42
43
44
45
46
47
48
 
# File 'lib/fluent/plugin/kafka_plugin_util.rb', line 39

def start
  super

  # This is bad point here but easy to fix for all kafka plugins
  unless log.respond_to?(:formatter)
    def log.formatter
      Fluent::KafkaPluginUtil::SSLSettings::DummyFormatter
    end
  end
end