Class: Fluent::ElasticsearchClusterOutput

Inherits:
BufferedOutput
  • Object
show all
Includes:
SetTagKeyMixin
Defined in:
lib/fluent/plugin/out_elasticsearch_cluster.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ElasticsearchClusterOutput 



25
26
27
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 25

def initialize
  super
end

Instance Method Details

#configure(conf) ⇒ Object 



29
30
31
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 29

def configure(conf)
  super
end

#format(tag, time, record) ⇒ Object 



47
48
49
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 47

def format(tag, time, record)
  [tag, time, record].to_msgpack
end

#get_hosts ⇒ Object 



39
40
41
42
43
44
45
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 39

def get_hosts
  if @hosts
      @hosts.split(',').map {|x| x.strip}.compact
   else
     ["#{@host}:#{@port}"]
   end
end

#send(data) ⇒ Object 



96
97
98
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 96

def send(data)
  @es.bulk body: data
end

#shutdown ⇒ Object 



51
52
53
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 51

def shutdown
  super
end

#start ⇒ Object 



33
34
35
36
37
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 33

def start
  super
  @es = Elasticsearch::Client.new :hosts => get_hosts, :reload_connections => true, :adapter => :patron, :retry_on_failure => 5
  raise "Can not reach Elasticsearch cluster (#{@host}:#{@port})!" unless @es.ping
end

#write(chunk) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/fluent/plugin/out_elasticsearch_cluster.rb', line 55

def write(chunk)
  bulk_message = []

  chunk.msgpack_each do |tag, time, record|
    if @logstash_format
      record.merge!({"@timestamp" => Time.at(time).to_datetime.to_s})
      if @utc_index
        target_index = "#{@logstash_prefix}-#{Time.at(time).getutc.strftime("#{@logstash_dateformat}")}"
      else
        target_index = "#{@logstash_prefix}-#{Time.at(time).strftime("#{@logstash_dateformat}")}"
      end
    else
      target_index = @index_name
    end

    if @include_tag_key
      record.merge!(@tag_key => tag)
    end

    meta = { "index" => {"_index" => target_index, "_type" => type_name} }
    if @id_key && record[@id_key]
      meta['index']['_id'] = record[@id_key]
    end

    if @parent_key && record[@parent_key]
      meta['index']['_parent'] = record[@parent_key]
    end

    bulk_message << meta
    bulk_message << record

    if bulk_message.size >= @flush_size
      send(bulk_message)
      bulk_message.clear
    end
  end

  send(bulk_message) unless bulk_message.empty?
  bulk_message.clear
end