|
1 | | -# encoding: utf-8 |
2 | | -require "logstash/filters/base" |
3 | | -require "logstash/namespace" |
4 | | - |
5 | | -# A general search and replace tool which queries replacement values from a redis instance. |
6 | | -# |
7 | | -# This is actually a redis version of a translate plugin. <https://www.elastic.co/guide/en/logstash/current/plugins-filters-translate.html> |
8 | | -# |
9 | | -# Operationally, if the event field specified in the "field" configuration |
10 | | -# matches the EXACT contents of a redis key, the field's value will be substituted |
11 | | -# with the matched key's value from the redis GET <key> command. |
12 | | -# |
13 | | -# By default, the redis filter will replace the contents of the |
14 | | -# matching event field (in-place). However, by using the "destination" |
15 | | -# configuration item, you may also specify a target event field to |
16 | | -# populate with the new translated value. |
17 | | -# |
18 | | -# Alternatively, for simple string search and replacements for just a few values |
19 | | -# you might consider using the gsub function of the mutate filter. |
20 | | - |
| 1 | +# A custom Logstash filter plugin to enrich events using data fetched from Redis. |
21 | 2 | class LogStash::Filters::Redis < LogStash::Filters::Base |
22 | | - |
23 | 3 | config_name "redis" |
24 | 4 |
|
25 | | - # The hostname of your Redis server. |
| 5 | + # Redis connection settings |
26 | 6 | config :host, :validate => :string, :default => "127.0.0.1" |
27 | | - |
28 | | - # The port to connect on. |
29 | 7 | config :port, :validate => :number, :default => 6379 |
30 | | - |
31 | | - # Password to authenticate with. There is no authentication by default. |
32 | 8 | config :password, :validate => :password |
33 | | - |
34 | | - # The Redis database number. |
35 | 9 | config :db, :validate => :number, :default => 0 |
36 | | - |
37 | | - # The name of the logstash event field containing the value to be compared for a |
38 | | - # match by the translate filter (e.g. "message", "host", "response_code"). |
39 | | - # |
40 | | - # If this field is an array, only the first value will be used. |
41 | | - config :field, :validate => :string, :required => true |
42 | | - |
43 | | - # If the destination (or target) field already exists, this configuration item specifies |
44 | | - # whether the filter should skip translation (default) or overwrite the target field |
45 | | - # value with the new translation value. |
46 | | - config :override, :validate => :boolean, :default => false |
47 | | - |
48 | | - # The destination field you wish to populate with the translated code. The default |
49 | | - # is a field named "redis". Set this to the same value as source if you want |
50 | | - # to do a substitution, in this case filter will allways succeed. This will clobber |
51 | | - # the old value of the source field! |
52 | | - config :destination, :validate => :string, :default => "redis" |
53 | | - |
54 | | - # In case no translation occurs in the event (no matches), this will add a default |
55 | | - # translation string, which will always populate "field", if the match failed. |
56 | | - # |
57 | | - # For example, if we have configured `fallback => "no match"`, using this dictionary: |
58 | | - # |
59 | | - # foo: bar |
60 | | - # |
61 | | - # Then, if logstash received an event with the field `foo` set to "bar", the destination |
62 | | - # field would be set to "bar". However, if logstash received an event with `foo` set to "nope", |
63 | | - # then the destination field would still be populated, but with the value of "no match". |
64 | | - config :fallback, :validate => :string |
65 | 10 |
|
66 | | - # Connection timeout |
67 | | - config :timeout, :validate => :number, :required => false, :default => 5 |
| 11 | + # Event processing options |
| 12 | + config :field, :validate => :string, :required => true # Field whose value will be used as the Redis key |
| 13 | + config :override, :validate => :boolean, :default => false # Whether to overwrite the destination field if it already exists |
| 14 | + config :destination, :validate => :string, :default => "redis" # Where to store the retrieved value in the event |
| 15 | + config :fallback, :validate => :string # Value to set if lookup fails or key doesn't exist |
| 16 | + config :timeout, :validate => :number, :required => false, :default => 5 # Redis connection timeout |
68 | 17 |
|
69 | 18 | public |
70 | 19 | def register |
71 | 20 | require 'redis' |
72 | 21 | require 'json' |
73 | 22 | @redis = nil |
74 | | - end # def register |
| 23 | + end |
75 | 24 |
|
76 | 25 | public |
77 | 26 | def filter(event) |
| 27 | + # Skip processing if the target field is missing or if destination exists and override is false |
78 | 28 | return unless event.include?(@field) |
79 | | - return if event.include?(@destination) and not @override |
| 29 | + return if event.include?(@destination) && !@override |
80 | 30 |
|
| 31 | + # Resolve source key from event (handle both string and array values) |
81 | 32 | source = event.get(@field).is_a?(Array) ? event.get(@field).first.to_s : event.get(@field).to_s |
82 | | - @redis ||= connect |
83 | | - val = @redis.get(source) |
84 | | - if val |
85 | | - begin |
86 | | - event.set(@destination, JSON.parse(val)) |
87 | | - rescue JSON::ParserError => e |
88 | | - event.set(@destination, val) |
| 33 | + |
| 34 | + begin |
| 35 | + @redis ||= connect |
| 36 | + type = @redis.type(source) |
| 37 | + |
| 38 | + case type |
| 39 | + when "string" |
| 40 | + value = @redis.get(source) |
| 41 | + event.set(@destination, value) if value |
| 42 | + |
| 43 | + when "hash" |
| 44 | + hash = @redis.hgetall(source) |
| 45 | + hash.each { |k, v| event.set("#{@destination}[#{k}]", v) } unless hash.empty? |
| 46 | + |
| 47 | + when "list" |
| 48 | + list = @redis.lrange(source, 0, -1) |
| 49 | + event.set(@destination, list) unless list.empty? |
| 50 | + |
| 51 | + when "set" |
| 52 | + set = @redis.smembers(source) |
| 53 | + event.set(@destination, set) unless set.empty? |
| 54 | + |
| 55 | + when "zset" |
| 56 | + zset = @redis.zrange(source, 0, -1, with_scores: true) |
| 57 | + event.set(@destination, zset) unless zset.empty? |
| 58 | + |
| 59 | + else |
| 60 | + # Unsupported or nonexistent key type, use fallback if available |
| 61 | + event.set(@destination, @fallback) if @fallback |
89 | 62 | end |
90 | | - elsif @fallback |
91 | | - event.set(@destination, @fallback) |
| 63 | + |
| 64 | + rescue => e |
| 65 | + # On error (connection, lookup, etc.), log and optionally set fallback |
| 66 | + @logger.warn("Redis lookup failed", :error => e.message) |
| 67 | + event.set(@destination, @fallback) if @fallback |
92 | 68 | end |
93 | | - |
94 | | - # filter_matched should go in the last line of our successful code |
| 69 | + |
95 | 70 | filter_matched(event) |
96 | | - end # def filter |
| 71 | + end |
97 | 72 |
|
98 | 73 | private |
| 74 | + # Establish a new Redis connection using configured options |
99 | 75 | def connect |
100 | | - Redis.new( |
101 | | - :host => @host, |
102 | | - :port => @port, |
103 | | - :timeout => @timeout, |
104 | | - :db => @db, |
105 | | - :password => @password.nil? ? nil : @password.value |
106 | | - ) |
107 | | - end #def connect |
108 | | -end # class LogStash::Filters::Redis |
| 76 | + Redis.new( |
| 77 | + host: @host, |
| 78 | + port: @port, |
| 79 | + timeout: @timeout, |
| 80 | + db: @db, |
| 81 | + password: @password.nil? ? nil : @password.value |
| 82 | + ) |
| 83 | + end |
| 84 | +end |
0 commit comments