Class: Fluent::CloudSearchOutput

Inherits:
BufferedOutput
  • Object
show all
Defined in:
lib/fluent/plugin/out_cloudsearch.rb

Constant Summary collapse

MAX_SIZE_LIMIT =

message packをJSONにした時に5MBを超えないように

4.5 * 1024 * 1024

Instance Method Summary collapse

Constructor Details

#initialize ⇒ CloudSearchOutput

Returns a new instance of CloudSearchOutput.



18
19
20
21
22
23
 
# File 'lib/fluent/plugin/out_cloudsearch.rb', line 18

def initialize
  super

  require 'aws-sdk'
  require 'json'
end

Instance Method Details

#configure(conf) ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/fluent/plugin/out_cloudsearch.rb', line 25

def configure(conf)

  # override config. (config_set_default can't override it)
  conf['buffer_chunk_limit'] ||= MAX_SIZE_LIMIT

  super

  unless @endpoint
    raise ConfigError, "'endpoint' parameter is required"
  end
  if @buffer.buffer_chunk_limit > MAX_SIZE_LIMIT
    raise ConfigError, "buffer_chunk_limit must be less than #{MAX_SIZE_LIMIT}"
  end
end

#format(tag, time, record) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/fluent/plugin/out_cloudsearch.rb', line 53

def format(tag, time, record)
  if !record.key?('id') then
    log.warn "id is required #{record.to_s}"
    return ''
  elsif !record.key?('type') then
    log.warn "type is required #{record.to_s}"
    return ''
  elsif record['type'] == 'add' then
    if !record.key?('fields') then
        log.warn "fields is required when type is add. #{record.to_s}"
        return ''
    end
  elsif record['type'] != 'delete' then
    log.warn "type is add or delete #{record.to_s}"
    return ''
  end

  "#{record.to_json},"
end

#setup_credentials ⇒ Object

docs.aws.amazon.com/sdkforruby/api/Aws/CloudSearchDomain/Client.html#initialize-instance_method



84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/fluent/plugin/out_cloudsearch.rb', line 84

def setup_credentials
  options = {}
  if @access_key_id && @secret_access_key
    options[:credentials] = Aws::Credentials.new(@access_key_id, @secret_access_key)
  elsif @profile_name
    options[:credentials] = Aws::SharedCredentials.new(
      :profile_name => @profile_name
    )
  end
  options
end

#shutdown ⇒ Object 



48
49
50
51
 
# File 'lib/fluent/plugin/out_cloudsearch.rb', line 48

def shutdown
  super

end

#start ⇒ Object 



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

def start
  super
  options = setup_credentials
  options[:endpoint] = @endpoint
  options[:region] = @region if @region
  @client = Aws::CloudSearchDomain::Client.new(options)
end

#write(chunk) ⇒ Object 



73
74
75
76
77
78
79
80
81
 
# File 'lib/fluent/plugin/out_cloudsearch.rb', line 73

def write(chunk)
  documents = '['
  documents << chunk.read.chop  # chop last ','
  documents << ']'
  resp = @client.upload_documents(
    :documents => documents,
    :content_type => "application/json"
  )
end