Class: LogStash::Outputs::Swift

Inherits:
Base
  • Object
show all
Defined in:
lib/logstash/outputs/swift.rb,
lib/logstash/outputs/swift/uploader.rb,
lib/logstash/outputs/swift/path_validator.rb,
lib/logstash/outputs/swift/temporary_file.rb,
lib/logstash/outputs/swift/file_repository.rb,
lib/logstash/outputs/swift/size_rotation_policy.rb,
lib/logstash/outputs/swift/time_rotation_policy.rb,
lib/logstash/outputs/swift/temporary_file_factory.rb,
lib/logstash/outputs/swift/writable_directory_validator.rb,
lib/logstash/outputs/swift/size_and_time_rotation_policy.rb,
lib/logstash/outputs/swift/write_container_permission_validator.rb

Overview

INFORMATION:

This plugin batches and uploads logstash events into Openstack Swift.

Requirements:

  • Amazon S3 Bucket and S3 Access Permissions (Typically access_key_id and secret_access_key)

  • S3 PutObject permission

Swift outputs create temporary files into the OS’ temporary directory, you can specify where to save them using the ‘temporary_directory` option.

Swift output files have the following format

ls.s3.312bc026-2f5d-49bc-ae9f-5940cf4ad9a6.2013-04-18T10.00.tag_hello.part0.txt

|======= | ls.s3 | indicate logstash plugin s3 | | 312bc026-2f5d-49bc-ae9f-5940cf4ad9a6 | a new, random uuid per file. | | 2013-04-18T10.00 | represents the time whenever you specify time_file. | | tag_hello | this indicates the event’s tag. | | part0 | this means if you indicate size_file then it will generate more parts if you file.size > size_file. When a file is full it will be pushed to the bucket and then deleted from the temporary directory. If a file is empty, it is simply deleted. Empty files will not be pushed | |=======

Crash Recovery:

  • This plugin will recover and upload temporary log files after crash/abnormal termination when using ‘restore` set to true

Note regarding time_file and size_file

:

Both time_file and size_file settings can trigger a log “file rotation” A log rotation pushes the current log “part” to s3 and deleted from local temporary storage.

If you specify BOTH size_file and time_file then it will create file for each tag (if specified). When EITHER time_file minutes have elapsed OR log file size > size_file, a log rotation is triggered.

If you ONLY specify time_file but NOT file_size, one file for each tag (if specified) will be created. When time_file minutes elapses, a log rotation will be triggered.

If you ONLY specify size_file, but NOT time_file, one files for each tag (if specified) will be created. When size of log file part > size_file, a log rotation will be triggered.

If NEITHER size_file nor time_file is specified, ONLY one file for each tag (if specified) will be created. WARNING: Since no log rotation is triggered, S3 Upload will only occur when logstash restarts.

#### Usage: This is an example of logstash config:

source,ruby

output {

swift {
  username => "crazy_name"                 (required)
  api_key => "crazy_key"                   (required)
  auth_url => "https://auth_tokens"        (required)
  project_name => "crazy project name"     (required)
  domain_name => "crazy domain name"       (required)
  container => "your_bucket_container"     (required)
}

Defined Under Namespace

Classes: FileRepository, PathValidator, SizeAndTimeRotationPolicy, SizeRotationPolicy, TemporaryFile, TemporaryFileFactory, TimeRotationPolicy, Uploader, WritableDirectoryValidator, WriteContainerPermissionValidator

Constant Summary collapse

PREFIX_KEY_NORMALIZE_CHARACTER =

include LogStash::PluginMixins::AwsConfig::V2

"_"
PERIODIC_CHECK_INTERVAL_IN_SECONDS = 
15
CRASH_RECOVERY_THREADPOOL = 
Concurrent::ThreadPoolExecutor.new({
  :min_threads => 1,
  :max_threads => 2,
  :fallback_policy => :caller_runs
})

Instance Method Summary collapse

Instance Method Details

#close ⇒ Object 



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/logstash/outputs/swift.rb', line 240

def close
  stop_periodic_check if @rotation.needs_periodic?

  @logger.debug("Uploading current workspace")

  # The plugin has stopped receiving new events, but we still have
  # data on disk, lets make sure it get to S3.
  # If Logstash get interrupted, the `restore_from_crash` (when set to true) method will pickup
  # the content in the temporary directly and upload it.
  # This will block the shutdown until all upload are done or the use force quit.
  @file_repository.each_files do |file|
    upload_file(file)
  end

  @file_repository.shutdown

  @uploader.stop # wait until all the current upload are complete
  @crash_uploader.stop if @restore # we might have still work to do for recovery so wait until we are done
end

#multi_receive_encoded(events_and_encoded) ⇒ Object 



216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/logstash/outputs/swift.rb', line 216

def multi_receive_encoded(events_and_encoded)
  puts 'milti rrecei'
  prefix_written_to = Set.new

  events_and_encoded.each do |event, encoded|
    prefix_key = normalize_key(event.sprintf(@prefix))
    prefix_written_to << prefix_key

    begin
      puts 'begin'
      @file_repository.get_file(prefix_key) { |file| file.write(encoded) }
      # The output should stop accepting new events coming in, since it cannot do anything with them anymore.
      # Log the error and rethrow it.
    rescue Errno::ENOSPC => e
      @logger.error("S3: No space left in temporary directory", :temporary_directory => @temporary_directory)
      raise e
    end
  end

  puts 'done'
  # Groups IO calls to optimize fstat checks
  rotate_if_needed(prefix_written_to)
end

#normalize_key(prefix_key) ⇒ Object 



260
261
262
 
# File 'lib/logstash/outputs/swift.rb', line 260

def normalize_key(prefix_key)
  prefix_key.gsub(PathValidator.matches_re, PREFIX_KEY_NORMALIZE_CHARACTER)
end

#register ⇒ Object 



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/logstash/outputs/swift.rb', line 174

def register
  # I've move the validation of the items into custom classes
  # to prepare for the new config validation that will be part of the core so the core can
  # be moved easily.
  unless @prefix.empty?
    if !PathValidator.valid?(prefix)
      raise LogStash::ConfigurationError, "Prefix must not contains: #{PathValidator::INVALID_CHARACTERS}"
    end
  end

  if !WritableDirectoryValidator.valid?(@temporary_directory)
    raise LogStash::ConfigurationError, "Logstash must have the permissions to write to the temporary directory: #{@temporary_directory}"
  end

  if @validate_credentials_on_root_bucket && !WriteContainerPermissionValidator.new(@logger).valid?(container_resource)
    raise LogStash::ConfigurationError, "Logstash must have the privileges to write to root container `#{@container}`, check your credentials or your permissions."
  end

  if @time_file.nil? && @size_file.nil? || @size_file == 0 && @time_file == 0
    raise LogStash::ConfigurationError, "The S3 plugin must have at least one of time_file or size_file set to a value greater than 0"
  end

  @file_repository = FileRepository.new(@tags, @encoding, @temporary_directory)

  @rotation = rotation_strategy

  executor = Concurrent::ThreadPoolExecutor.new({ :min_threads => 1,
                                                  :max_threads => @upload_workers_count,
                                                  :max_queue => @upload_queue_size,
                                                  :fallback_policy => :caller_runs })

  @uploader = Uploader.new(container_resource, @logger, executor)

  # Restoring from crash will use a new threadpool to slowly recover
  # New events should have more priority.
  restore_from_crash if @restore

  # If we need time based rotation we need to do periodic check on the file
  # to take care of file that were not updated recently
  start_periodic_check if @rotation.needs_periodic?
end

#restore ⇒ Object

IMPORTANT: if you use multiple instance of s3, you should specify on one of them the “restore=> true” and on the others “restore => false”. This is hack for not destroy the new files after restoring the initial files. If you do not specify “restore => true” when logstash crashes or is restarted, the files are not sent into the bucket, for example if you have single Instance.



118
 
# File 'lib/logstash/outputs/swift.rb', line 118

config :restore, :validate => :boolean, :default => true