Adding feature to support dynamic prefix

CHANGELOG.md

@@ -1,3 +1,5 @@
+## 2.0.6
+ - Feature to support dynamic prefix
 ## 2.0.5
  - Support signature_version option for v4 S3 keys
 ## 2.0.4
@@ -14,4 +16,4 @@
 
 # 1.0.1
 - Fix a synchronization issue when doing file rotation and checking the size of the current file
-- Fix an issue with synchronization when shutting down the plugin and closing the current temp file
+- Fix an issue with synchronization when shutting down the plugin and closing the current temp file

CONTRIBUTORS

@@ -10,8 +10,9 @@ Contributors:
 * Nick Ethier (nickethier)
 * Pier-Hugues Pellerin (ph)
 * Richard Pijnenburg (electrical)
+* Sergio Díaz (sdiazb)
 
 Note: If you've sent us patches, bug reports, or otherwise contributed to
 Logstash, and you aren't on the list above and want to be, please let us know
 and we'll make sure you're here. Contributions from folks like you are what make
-open source awesome.
+open source awesome.

lib/logstash/outputs/s3.rb

@@ -104,6 +104,10 @@ class LogStash::Outputs::S3 < LogStash::Outputs::Base
   config :temporary_directory, :validate => :string, :default => File.join(Dir.tmpdir, "logstash")
 
   # Specify a prefix to the uploaded filename, this can simulate directories on S3
+  # One may also utilize the path option for date-based log
+  # rotation. This will use the event timestamp.
+  # E.g.: `prefix => "test/%Y%m%d/"` to create
+  # `test/20160216/`
   config :prefix, :validate => :string, :default => ''
 
   # Specify how many workers to use to upload the files to S3
@@ -128,6 +132,8 @@ class LogStash::Outputs::S3 < LogStash::Outputs::Base
   attr_reader :page_counter
   attr_reader :s3
 
+  attr_accessor :original_prefix
+
   def aws_s3_config
     @logger.info("Registering s3 output", :bucket => @bucket, :endpoint_region => @region)
     @s3 = AWS::S3.new(full_options)
@@ -165,6 +171,8 @@ def write_on_bucket(file)
     # find and use the bucket
     bucket = @s3.buckets[@bucket]
 
+    build_prefix(Time.now)
+
     remote_filename = "#{@prefix}#{File.basename(file)}"
 
     @logger.debug("S3: ready to write file in bucket", :remote_filename => remote_filename, :bucket => @bucket)
@@ -217,6 +225,10 @@ def register
       raise LogStash::ConfigurationError, "S3: prefix contains invalid characters"
     end
 
+    if @original_prefix.nil?
+      @original_prefix = @prefix
+    end
+
     if !Dir.exist?(@temporary_directory)
       FileUtils.mkdir_p(@temporary_directory)
     end
@@ -303,7 +315,6 @@ def get_temporary_filename(page_counter = 0)
 
   public
   def receive(event)
-
     @codec.encode(event)
   end
 
@@ -343,6 +354,11 @@ def close
     end
   end
 
+  public
+  def build_prefix(datetime)
+    @prefix = datetime.strftime(@original_prefix)
+  end
+
   private
   def shutdown_upload_workers
     @logger.debug("S3: Gracefully shutdown the upload workers")

logstash-output-s3.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-output-s3'
-  s.version         = '2.0.5'
+  s.version         = '2.0.6'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "This plugin was created for store the logstash's events into Amazon Simple Storage Service (Amazon S3)"
   s.description     = "This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program"
@@ -26,4 +26,4 @@ Gem::Specification.new do |s|
   s.add_development_dependency 'logstash-devutils'
   s.add_development_dependency 'logstash-input-generator'
   s.add_development_dependency 'logstash-codec-line'
-end
+end

spec/outputs/s3_spec.rb

@@ -206,7 +206,7 @@
 
       after(:each) do
         s3.close
-        tmp.close 
+        tmp.close
         tmp.unlink
       end
 
@@ -351,4 +351,20 @@
       end
     end
   end
+
+  describe "#build_prefix" do
+    it "should render event fields and date in prefix" do
+      config = {
+        "prefix" => "foo/bar/%Y%m%d/"
+      }
+
+      s3 = LogStash::Outputs::S3.new(config)
+
+      s3.original_prefix = s3.prefix
+
+      s3.build_prefix(Time.new(2016, 02, 17))
+
+      expect(s3.prefix).to eql('foo/bar/20160217/')
+    end
+  end
 end