Class: Fluent::Plugin::FileBuffer
- Includes:
- SystemConfig::Mixin
- Defined in:
- lib/fluent/plugin/buf_file.rb
Constant Summary collapse
- DEFAULT_CHUNK_LIMIT_SIZE =
256MB
256 * 1024 * 1024
- DEFAULT_TOTAL_LIMIT_SIZE =
64GB, same with v0.12 (TimeSlicedOutput + buf_file)
64 * 1024 * 1024 * 1024
- DIR_PERMISSION =
0755
- @@buffer_paths =
TODO: Buffer plugin cannot handle symlinks because new API @stage has many writing buffer chunks
re-implement this feature on out_file, w/ enqueue_chunk(or generate_chunk) hook + chunk.path
attr_accessor :symlink_path
{}
Constants inherited from Buffer
Buffer::DEFAULT_CHUNK_FULL_THRESHOLD, Buffer::MINIMUM_APPEND_ATTEMPT_RECORDS
Constants included from Configurable
Configurable::CONFIG_TYPE_REGISTRY
Instance Attribute Summary
Attributes inherited from Buffer
#dequeued, #queue, #queue_size, #queued_num, #stage, #stage_size
Instance Method Summary collapse
- #buffer_path_for_test? ⇒ Boolean
- #configure(conf) ⇒ Object
- #generate_chunk(metadata) ⇒ Object
-
#initialize ⇒ FileBuffer
constructor
A new instance of FileBuffer.
- #persistent? ⇒ Boolean
- #resume ⇒ Object
- #start ⇒ Object
Methods included from SystemConfig::Mixin
#system_config, #system_config_override
Methods inherited from Buffer
#add_metadata, #chunk_size_full?, #chunk_size_over?, #clear_queue!, #close, #dequeue_chunk, #enqueue_all, #enqueue_chunk, #enqueue_unstaged_chunk, #metadata, #metadata_list, #new_metadata, #purge_chunk, #queued?, #queued_records, #storable?, #takeback_chunk, #terminate, #write, #write_once, #write_step_by_step
Methods included from UniqueId::Mixin
#dump_unique_id_hex, #generate_unique_id
Methods included from OwnedByMixin
Methods inherited from Base
#after_shutdown, #after_shutdown?, #after_start, #after_started?, #before_shutdown, #before_shutdown?, #close, #closed?, #configured?, #has_router?, #inspect, #shutdown, #shutdown?, #started?, #stop, #stopped?, #terminate, #terminated?
Methods included from Configurable
#config, included, lookup_type, register_type
Constructor Details
#initialize ⇒ FileBuffer
Returns a new instance of FileBuffer.
51 52 53 54 |
# File 'lib/fluent/plugin/buf_file.rb', line 51 def initialize super @symlink_path = nil end |
Instance Method Details
#buffer_path_for_test? ⇒ Boolean
95 96 97 98 99 100 101 102 103 104 105 |
# File 'lib/fluent/plugin/buf_file.rb', line 95 def buffer_path_for_test? caller_locations.each do |location| # Thread::Backtrace::Location#path returns base filename or absolute path. # #absolute_path returns absolute_path always. # https://bugs.ruby-lang.org/issues/12159 if location.absolute_path =~ /\/test_[^\/]+\.rb$/ # location.path =~ /test_.+\.rb$/ return true end end false end |
#configure(conf) ⇒ Object
56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
# File 'lib/fluent/plugin/buf_file.rb', line 56 def configure(conf) super type_of_owner = Plugin.lookup_type_from_class(@_owner.class) if @@buffer_paths.has_key?(@path) && !buffer_path_for_test? type_using_this_path = @@buffer_paths[@path] raise ConfigError, "Other '#{type_using_this_path}' plugin already use same buffer path: type = #{type_of_owner}, buffer path = #{@path}" end @@buffer_paths[@path] = type_of_owner # TODO: create buffer path with plugin_id, under directory specified by system config if File.exist?(@path) if File.directory?(@path) @path = File.join(@path, 'buffer.*.log') elsif File.basename(@path).include?('.*.') # valid path (buffer.*.log will be ignored) elsif File.basename(@path).end_with?('.*') @path = @path + '.log' else # existing file will be ignored @path = @path + '.*.log' end else # path doesn't exist if File.basename(@path).include?('.*.') # valid path elsif File.basename(@path).end_with?('.*') @path = @path + '.log' else # path is handled as directory, and it will be created at #start @path = File.join(@path, 'buffer.*.log') end end unless @dir_permission @dir_permission = system_config. || DIR_PERMISSION end end |
#generate_chunk(metadata) ⇒ Object
144 145 146 147 148 149 150 151 |
# File 'lib/fluent/plugin/buf_file.rb', line 144 def generate_chunk() # FileChunk generates real path with unique_id if @file_permission Fluent::Plugin::Buffer::FileChunk.new(, @path, :create, perm: @file_permission) else Fluent::Plugin::Buffer::FileChunk.new(, @path, :create) end end |
#persistent? ⇒ Boolean
113 114 115 |
# File 'lib/fluent/plugin/buf_file.rb', line 113 def persistent? true end |
#resume ⇒ Object
117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
# File 'lib/fluent/plugin/buf_file.rb', line 117 def resume stage = {} queue = [] Dir.glob(@path) do |path| m = () # this metadata will be overwritten by resuming .meta file content # so it should not added into @metadata_list for now mode = Fluent::Plugin::Buffer::FileChunk.assume_chunk_state(path) if mode == :unknown log.debug "uknown state chunk found", path: path next end chunk = Fluent::Plugin::Buffer::FileChunk.new(m, path, mode) # file chunk resumes contents of metadata case chunk.state when :staged stage[chunk.] = chunk when :queued queue << chunk end end queue.sort_by!{ |chunk| chunk.modified_at } return stage, queue end |
#start ⇒ Object
107 108 109 110 111 |
# File 'lib/fluent/plugin/buf_file.rb', line 107 def start FileUtils.mkdir_p File.dirname(@path), mode: @dir_permission super end |