Class: Fluent::Plugin::RedisStoreOutput
- Inherits:
-
Output
- Object
- Output
- Fluent::Plugin::RedisStoreOutput
- Defined in:
- lib/fluent/plugin/out_redis_store.rb
Constant Summary collapse
- DEFAULT_BUFFER_TYPE =
"memory"
Instance Method Summary collapse
- #configure(conf) ⇒ Object
- #format(tag, time, record) ⇒ Object
- #formatted_to_msgpack_binary? ⇒ Boolean
- #generate_ltrim_script(key, maxlen, order) ⇒ Object
- #generate_zremrangebyrank_script(key, maxlen, order) ⇒ Object
- #get_key_from(record) ⇒ Object
- #get_score_from(record, time) ⇒ Object
- #get_value_from(record) ⇒ Object
-
#initialize ⇒ RedisStoreOutput
constructor
A new instance of RedisStoreOutput.
- #multi_workers_ready? ⇒ Boolean
- #operation_for_list(record) ⇒ Object
- #operation_for_publish(record) ⇒ Object
- #operation_for_set(record) ⇒ Object
- #operation_for_string(record) ⇒ Object
- #operation_for_zset(record, time) ⇒ Object
- #set_key_expire(key) ⇒ Object
- #shutdown ⇒ Object
- #start ⇒ Object
- #traverse(data, key) ⇒ Object
- #write(chunk) ⇒ Object
Constructor Details
#initialize ⇒ RedisStoreOutput
Returns a new instance of RedisStoreOutput.
41 42 43 44 45 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 41 def initialize super require 'redis' unless defined?(Redis) == 'constant' require 'msgpack' end |
Instance Method Details
#configure(conf) ⇒ Object
47 48 49 50 51 52 53 54 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 47 def configure(conf) compat_parameters_convert(conf, :buffer) super if @key_path == nil and @key == nil raise Fluent::ConfigError, "either key_path or key is required" end end |
#format(tag, time, record) ⇒ Object
72 73 74 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 72 def format(tag, time, record) [tag, time.to_f, record].to_msgpack end |
#formatted_to_msgpack_binary? ⇒ Boolean
76 77 78 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 76 def formatted_to_msgpack_binary? true end |
#generate_ltrim_script(key, maxlen, order) ⇒ Object
205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 205 def generate_ltrim_script(key, maxlen, order) script = "local key = '" + key.to_s + "'\n" script += "local maxlen = " + maxlen.to_s + "\n" script += "local order ='" + order.to_s + "'\n" script += "local len = tonumber(redis.call('LLEN', key))\n" script += "if len > maxlen then\n" script += " if order == 'asc' then\n" script += " local l = len - maxlen\n" script += " return redis.call('LTRIM', key, l, -1)\n" script += " else\n" script += " return redis.call('LTRIM', key, 0, maxlen - 1)\n" script += " end\n" script += "end\n" return script end |
#generate_zremrangebyrank_script(key, maxlen, order) ⇒ Object
187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 187 def generate_zremrangebyrank_script(key, maxlen, order) script = "local key = '" + key.to_s + "'\n" script += "local maxlen = " + maxlen.to_s + "\n" script += "local order ='" + order.to_s + "'\n" script += "local len = tonumber(redis.call('ZCOUNT', key, '-inf', '+inf'))\n" script += "if len > maxlen then\n" script += " if order == 'asc' then\n" script += " local l = len - maxlen\n" script += " if l >= 0 then\n" script += " return redis.call('ZREMRANGEBYRANK', key, 0, l)\n" script += " end\n" script += " else\n" script += " return redis.call('ZREMRANGEBYRANK', key, maxlen, -1)\n" script += " end\n" script += "end\n" return script end |
#get_key_from(record) ⇒ Object
233 234 235 236 237 238 239 240 241 242 243 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 233 def get_key_from(record) if @key k = @key else k = traverse(record, @key_path).to_s end key = @key_prefix + k + @key_suffix raise Fluent::ConfigError, "key is empty" if key == '' key end |
#get_score_from(record, time) ⇒ Object
257 258 259 260 261 262 263 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 257 def get_score_from(record, time) if @score_path traverse(record, @score_path) else time end end |
#get_value_from(record) ⇒ Object
245 246 247 248 249 250 251 252 253 254 255 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 245 def get_value_from(record) value = traverse(record, @value_path) case @format_type when 'json' value.to_json when 'msgpack' value.to_msgpack else value end end |
#multi_workers_ready? ⇒ Boolean
80 81 82 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 80 def multi_workers_ready? true end |
#operation_for_list(record) ⇒ Object
150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 150 def operation_for_list(record) key = get_key_from(record) value = get_value_from(record) if @order == 'asc' @redis.rpush key, value else @redis.lpush key, value end @redis.ltrim key, 0, @max_list_length - 1 if 0 < @max_list_length set_key_expire key if 0 < @value_length script = generate_ltrim_script(key, @value_length, @order) @redis.eval script end end |
#operation_for_publish(record) ⇒ Object
177 178 179 180 181 182 183 184 185 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 177 def operation_for_publish(record) key = get_key_from(record) value = get_value_from(record) @redis.publish key, value # logstream is the stream used by our logforwarder to send # log messages to user defined syslog endpoints @redis.xadd('logstream', {data: value}, maxlen: 400000, approximate: true) end |
#operation_for_set(record) ⇒ Object
143 144 145 146 147 148 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 143 def operation_for_set(record) key = get_key_from(record) value = get_value_from(record) @redis.sadd key, value set_key_expire key end |
#operation_for_string(record) ⇒ Object
169 170 171 172 173 174 175 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 169 def operation_for_string(record) key = get_key_from(record) value = get_value_from(record) @redis.set key, value set_key_expire key end |
#operation_for_zset(record, time) ⇒ Object
118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 118 def operation_for_zset(record, time) key = get_key_from(record) value = get_value_from(record) score = get_score_from(record, time) if @collision_policy if @collision_policy == 'NX' @redis.zadd(key, score, value, :nx => true) elsif @collision_policy == 'XX' @redis.zadd(key, score, value, :xx => true) end else @redis.zadd(key, score, value) end set_key_expire key if 0 < @value_expire now = Time.now.to_i @redis.zremrangebyscore key , '-inf' , (now - @value_expire) end if 0 < @value_length script = generate_zremrangebyrank_script(key, @value_length, @order) @redis.eval script end end |
#set_key_expire(key) ⇒ Object
265 266 267 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 265 def set_key_expire(key) @redis.expire key, @key_expire if 0 < @key_expire end |
#shutdown ⇒ Object
67 68 69 70 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 67 def shutdown @redis.quit super end |
#start ⇒ Object
56 57 58 59 60 61 62 63 64 65 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 56 def start super if @path @redis = Redis.new(:path => @path, :username => @username, :password => @password, :timeout => @timeout, :thread_safe => true, :db => @db) else @redis = Redis.new(:host => @host, :port => @port, :username => @username, :password => @password, :timeout => @timeout, :thread_safe => true, :db => @db) end end |
#traverse(data, key) ⇒ Object
221 222 223 224 225 226 227 228 229 230 231 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 221 def traverse(data, key) val = data key.split('.').each{ |k| if val.has_key?(k) val = val[k] else return nil end } return val end |
#write(chunk) ⇒ Object
84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
# File 'lib/fluent/plugin/out_redis_store.rb', line 84 def write(chunk) @redis.pipelined { chunk.open { |io| begin MessagePack::Unpacker.new(io).each { || begin (_, time, record) = case @store_type when 'zset' operation_for_zset(record, time) when 'set' operation_for_set(record) when 'list' operation_for_list(record) when 'string' operation_for_string(record) when 'publish' operation_for_publish(record) end rescue NoMethodError => e puts e rescue Encoding::UndefinedConversionError => e log.error "Plugin error: " + e.to_s log.error "Original record: " + record.to_s puts e end } rescue EOFError # EOFError always occured when reached end of chunk. end } } end |