Skip to content

Add support for the LPUSH command as a list operation #58

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add support for the LPUSH command as a list operation #58

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions docs/index.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,10 @@ include::{include_path}/plugin_header.asciidoc[]

==== Description

This output will send events to a Redis queue using RPUSH.
The RPUSH command is supported in Redis v0.0.7+. Using
PUBLISH to a channel requires at least v1.3.8+.
This output will send events to a Redis queue using RPUSH
(or, optionally, LPUSH).
The RPUSH and LPUSH commands are supported in Redis v1.0.0+.
Using PUBLISH to a channel requires at least v1.3.8+.
While you may be able to make these Redis versions work,
the best performance and stability will be found in more
recent stable versions. Versions 2.6.0+ are recommended.
Expand Down
35 changes: 24 additions & 11 deletions lib/logstash/outputs/redis.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,10 @@
require "logstash/namespace"
require "stud/buffer"

# This output will send events to a Redis queue using RPUSH.
# The RPUSH command is supported in Redis v0.0.7+. Using
# PUBLISH to a channel requires at least v1.3.8+.
# This output will send events to a Redis queue using RPUSH
# or LPUSH.
# The RPUSH and LPUSH commands are supported in Redis v1.0.0+.
# Using PUBLISH to a channel requires at least v1.3.8+.
# While you may be able to make these Redis versions work,
# the best performance and stability will be found in more
# recent stable versions. Versions 2.6.0+ are recommended.
Expand Down Expand Up @@ -56,15 +57,19 @@ class LogStash::Outputs::Redis < LogStash::Outputs::Base
# valid here, for example `logstash-%{type}`.
config :key, :validate => :string, :required => true

# Either list or channel. If `redis_type` is list, then we will set
# RPUSH to key. If `redis_type` is channel, then we will PUBLISH to `key`.
# Either list or channel. If `redis_type` is list, then we will set RPUSH (or
# LPUSH) to key. If `redis_type` is channel, then we will PUBLISH to `key`.
config :data_type, :validate => [ "list", "channel" ], :required => true

# Set to true if you want Redis to batch up values and send 1 RPUSH command
# instead of one command per value to push on the list. Note that this only
# works with `data_type="list"` mode right now.
# Either RPUSH or LPUSH. Only relevant if `data_type` is list. Defines if
# elements shall be inserted at the head of the list (LPUSH) or the tail (RPUSH).
config :list_operation, :validate => ["RPUSH", "LPUSH"], :default => "RPUSH"

# Set to true if you want Redis to batch up values and send 1 RPUSH (or
# LPUSH) command instead of one command per value to push on the list.
# Note that this only works with `data_type="list"` mode right now.
#
# If true, we send an RPUSH every "batch_events" events or
# If true, we send an RPUSH (or LPUSH) every "batch_events" events or
# "batch_timeout" seconds (whichever comes first).
# Only supported for `data_type` is "list".
config :batch, :validate => :boolean, :default => false
Expand Down Expand Up @@ -154,7 +159,11 @@ def flush(events, key, close=false)
# we should not block due to congestion on close
# to support this Stud::Buffer#buffer_flush should pass here the :final boolean value.
congestion_check(key) unless close
@redis.rpush(key, events)
if @list_operation == 'LPUSH'
@redis.lpush(key, events)
else
@redis.rpush(key, events)
end
end
# called from Stud::Buffer#buffer_flush when an error occurs
def on_flush_error(e)
Expand Down Expand Up @@ -219,7 +228,11 @@ def send_to_redis(event, payload)
@redis ||= connect
if @data_type == 'list'
congestion_check(key)
@redis.rpush(key, payload)
if @list_operation == 'LPUSH'
@redis.lpush(key, payload)
else
@redis.rpush(key, payload)
end
else
@redis.publish(key, payload)
end
Expand Down
2 changes: 1 addition & 1 deletion logstash-output-redis.gemspec
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ Gem::Specification.new do |s|
s.name = 'logstash-output-redis'
s.version = '4.0.3'
s.licenses = ['Apache License (2.0)']
s.summary = "Sends events to a Redis queue using the `RPUSH` command"
s.summary = "Sends events to a Redis queue using the `RPUSH` or `LPUSH` command"
s.description = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
s.authors = ["Elastic"]
s.email = '[email protected]'
Expand Down