Class: Fluent::Plugin::MongoStorage
Instance Attribute Summary collapse
Instance Method Summary
collapse
#configure_logger, included
Methods included from MongoAuth
#authenticate
included
Constructor Details
Returns a new instance of MongoStorage.
53
54
55
56
57
58
59
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 53
def initialize
super
@client_options = {}
@collection_options = {capped: false}
@store = {}
end
|
Instance Attribute Details
#client_options ⇒ Object
Returns the value of attribute client_options.
51
52
53
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 51
def client_options
@client_options
end
|
#collection_options ⇒ Object
Returns the value of attribute collection_options.
51
52
53
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 51
def collection_options
@collection_options
end
|
#store ⇒ Object
11
12
13
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 11
def store
@store
end
|
Instance Method Details
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/storage_mongo.rb', line 61
def configure(conf)
super
unless @path
if conf && !conf.arg.empty?
@path = conf.arg
else
raise Fluent::ConfigError, "path or conf.arg for <storage> is required."
end
end
if conf.has_key?('capped') and Fluent::Config.bool_value(conf['capped'])
raise Fluent::ConfigError, "'capped_size' parameter is required on <storage> of Mongo storage" unless conf.has_key?('capped_size')
@collection_options[:capped] = true
@collection_options[:size] = Fluent::Config.size_value(conf['capped_size'])
@collection_options[:max] = Fluent::Config.size_value(conf['capped_max']) if conf.has_key?('capped_max')
end
@client_options[:write] = {j: @journaled}
@client_options[:write].merge!({w: @write_concern}) unless @write_concern.nil?
@client_options[:ssl] = @ssl
if @ssl
@client_options[:ssl_cert] = @ssl_cert
@client_options[:ssl_key] = @ssl_key
@client_options[:ssl_key_pass_phrase] = @ssl_key_pass_phrase
@client_options[:ssl_verify] = @ssl_verify
@client_options[:ssl_ca_cert] = @ssl_ca_cert
end
configure_logger(@mongo_log_level)
@client = client
@client = authenticate(@client)
end
|
#delete(key) ⇒ Object
138
139
140
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 138
def delete(key)
@store.delete(key.to_s)
end
|
#fetch(key, defval) ⇒ Object
130
131
132
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 130
def fetch(key, defval)
@store.fetch(key.to_s, defval)
end
|
#get(key) ⇒ Object
126
127
128
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 126
def get(key)
@store[key.to_s]
end
|
#load ⇒ Object
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 100
def load
begin
value = {}
documents = @client[format_collection_name(@collection)].find(_id: @path)
if documents.count >= 1
documents.each do |document|
value.merge!(document)
end
end
value.delete('_id')
unless value.is_a?(Hash)
log.error "broken content for plugin storage (Hash required: ignored)", type: json.class
log.debug "broken content", content: json_string
return
end
@store = value
rescue => e
log.error "failed to load data for plugin storage from mongo", path: @path, error: e
end
end
|
#multi_workers_ready? ⇒ Boolean
96
97
98
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 96
def multi_workers_ready?
true
end
|
#persistent_always? ⇒ Boolean
47
48
49
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 47
def persistent_always?
true
end
|
#put(key, value) ⇒ Object
134
135
136
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 134
def put(key, value)
@store[key.to_s] = value
end
|
#save ⇒ Object
122
123
124
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 122
def save
operate(format_collection_name(@collection), @store)
end
|
#update(key, &block) ⇒ Object
142
143
144
|
# File 'lib/fluent/plugin/storage_mongo.rb', line 142
def update(key, &block)
@store[key.to_s] = block.call(@store[key.to_s])
end
|