Class: BucketProcessor

Inherits:

Object

• Object
• BucketProcessor

Defined in:

lib/s3-object-processor.rb

Instance Method Summary

• #initialize(key_id, key_secret, bucket, options = {}, &callback) ⇒ BucketProcessor constructor

  A new instance of BucketProcessor.

• #run(prefix = nil) ⇒ Object

Constructor Details

#initialize(key_id, key_secret, bucket, options = {}, &callback) ⇒ BucketProcessor

Returns a new instance of BucketProcessor.

# File 'lib/s3-object-processor.rb', line 204

def initialize(key_id, key_secret, bucket, options = {}, &callback)
  protocol = options[:no_https] ? 'http' : 'https'
  port = options[:no_https] ? 80 : 443
  @log = options[:log] || Logger.new(STDERR)
  workers = options[:workers] || 10
  lister_fetch_size = options[:lister_fetch_size] || 200
  lister_backlog = options[:lister_backlog] || 1000
  max_keys = options[:max_keys]
  reporter_backlog = options[:reporter_backlog] || 1000
  reporter_summary_interval = options[:reporter_summary_interval] || 100
  reporter_average_contribution = options[:reporter_average_contribution] || 0.10
  custom_reports = options[:reports] || []
  @key_list = options[:key_list]

  s3 = RightAws::S3.new(key_id, key_secret, multi_thread: true, logger: @log, protocol: protocol, port: port)
  bucket = s3.bucket(bucket)

  @key_queue = SizedQueue.new(lister_backlog)

  @reporter = Reporter.new(reporter_backlog) do |reports|
    total_listed_keys = 0
    total_processed_keys = 0
    total_succeeded_keys = 0
    total_failed_keys = 0
    total_handled_keys = 0
    total_skipped_keys = 0
    total_nooped_keys = 0

    processed_avg = 0.0
    last_time = nil
    last_total = 0

    reports.each do |key, value|
      case key
      when :new_keys_count
        total_listed_keys += value
      when :processed_key
        total_processed_keys += 1
        if total_processed_keys % reporter_summary_interval == 0
          if last_time
            contribution = reporter_average_contribution
            new = (total_processed_keys - last_total).to_f / (Time.now.to_f - last_time)
            processed_avg = processed_avg * (1.0 - contribution) + new * contribution
          end
          last_time = Time.now.to_f
          last_total = total_processed_keys

          log_line = "-[%s]- processed %6d: failed: %6d (%6.2f %%) handled: %6d skipped: %6d (%6.2f %%)" % [
            value.to_s[0...2].ljust(2),
            total_processed_keys,
            total_failed_keys,
            total_failed_keys.to_f / total_processed_keys * 100,
            total_handled_keys,
            total_skipped_keys,
            total_skipped_keys.to_f / total_processed_keys * 100
          ]
          log_line << custom_reports.each_value.map{|v| ' ' + v.to_s}.join
          log_line << " [backlog: %4d] @ %.1f op/s" % [
            @key_queue.size,
            processed_avg
          ]

          @log.info log_line
        end
      when :succeeded_key
        total_succeeded_keys += 1
      when :failed_key
        key, error = *value
        @log.error "Key processing failed: #{key}: #{error.class.name}, #{error.message}"
        total_failed_keys += 1
      when :handled_key
        total_handled_keys += 1
      when :skipped_key
        total_skipped_keys += 1
      when :noop_key
        total_nooped_keys += 1
      else
        #@log.debug "custom report event: #{key}: #{value}"
        custom_reports[key].update(value)
      end
      #@log.debug("Report: #{key}: #{value}")
    end

    reports.on_finish do
      @log.info("total listed keys:                      #{total_listed_keys}")
      @log.info("total processed keys:                   #{total_processed_keys}")
      @log.info("total succeeded keys:                   #{total_succeeded_keys}")
      @log.info("total failed keys:                      #{total_failed_keys}")
      @log.info("total handled keys:                     #{total_handled_keys}")
      @log.info("total skipped keys:                     #{total_skipped_keys}")
      @log.info("total nooped keys:                      #{total_nooped_keys}")
      custom_reports.each_value do |report|
        @log.info report.final
      end
    end
  end

  # create lister
  @lister = if @key_list
    @log.info "processing #{@key_list.length} keys from list file"
    @lister = ListLister.new(bucket, @key_queue, max_keys)
  else
    @lister = Lister.new(bucket, @key_queue, lister_fetch_size, max_keys)
  end
  .on_keys_chunk do |keys_chunk|
    @log.debug "Got #{keys_chunk.length} new keys"
    @reporter.report(:new_keys_count, keys_chunk.length)
  end
  .on_finish do
    @log.debug "Done listing keys"
    # notify all workers that no more messages will be posted
    workers.times{ @key_queue << :end }
  end

  # create workers
  @log.info "Launching #{workers} workers"
  @workers = (1..workers).to_a.map do |worker_no|
    Worker.new(worker_no, @key_queue) do |key|
      @log.debug "Worker[#{worker_no}]: Processing key #{key}"
      yield bucket, key, @reporter
      @reporter.report :processed_key, key
      @reporter.report :succeeded_key, key
    end
    .on_error do |key, error|
      @reporter.report :processed_key, key
      @reporter.report :failed_key, [key, error]
    end
    .on_finish do
      @log.debug "Worker #{worker_no} done"
    end
  end
end

Instance Method Details

#run(prefix = nil) ⇒ Object

# File 'lib/s3-object-processor.rb', line 337

def run(prefix = nil)
  begin
    @reporter.run
    if @key_list
      @lister.run(@key_list)
    else
      @lister.run(prefix)
    end
    @workers.each(&:run)

    # wait for all to finish
    @workers.each(&:join)
    @log.info "All workers done"

    @lister.join
    @reporter.join
  rescue Interrupt => error
    @log.warn 'Interrupted'
    # flush thread waiting on queues
    @key_queue.max = 999999
    @reporter.join
  end
end