Class: TreasureData::Logger::TreasureDataLogger
- Inherits:
-
Fluent::Logger::LoggerBase
- Object
- Fluent::Logger::LoggerBase
- TreasureData::Logger::TreasureDataLogger
- Defined in:
- lib/td/logger/td_logger.rb
Defined Under Namespace
Classes: Buffer
Instance Attribute Summary collapse
-
#logger ⇒ Object
Returns the value of attribute logger.
Instance Method Summary collapse
- #close ⇒ Object
- #flush ⇒ Object
-
#initialize(tag_prefix, options = {}) ⇒ TreasureDataLogger
constructor
A new instance of TreasureDataLogger.
- #post_with_time(tag, record, time) ⇒ Object
- #upload_main ⇒ Object
Constructor Details
#initialize(tag_prefix, options = {}) ⇒ TreasureDataLogger
Returns a new instance of TreasureDataLogger.
6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 |
# File 'lib/td/logger/td_logger.rb', line 6 def initialize(tag_prefix, ={}) defaults = { :auto_create_table => false, } = defaults.merge!() @tag_prefix = tag_prefix @auto_create_table = !![:auto_create_table] apikey = [:apikey] unless apikey raise ArgumentError, ":apikey options is required" end debug = !![:debug] require 'thread' require 'stringio' require 'zlib' require 'msgpack' require 'json' require 'time' require 'net/http' require 'cgi' require 'logger' require 'td-client' @logger = ::Logger.new(STDERR) if debug @logger.level = ::Logger::DEBUG else @logger.level = ::Logger::INFO end @client = TreasureData::Client.new(apikey, :ssl => !![:use_ssl]) @mutex = Mutex.new @cond = ConditionVariable.new @map = {} # (db,table) => buffer:String @queue = [] @chunk_limit = 8*1024*1024 @queue_limit = 50 @flush_interval = 2 @max_flush_interval = 300 @retry_wait = 1.0 @retry_limit = 12 @finish = false @next_time = Time.now.to_i + @flush_interval @error_count = 0 # start thread when the first post() is called for # Unicorn and Passenger. @upload_thread = nil # The calling order of finalizer registered by define_finalizer is indeterminate, # so we should use at_exit instead for memory safety. at_exit { close } end |
Instance Attribute Details
#logger ⇒ Object
Returns the value of attribute logger.
68 69 70 |
# File 'lib/td/logger/td_logger.rb', line 68 def logger @logger end |
Instance Method Details
#close ⇒ Object
70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
# File 'lib/td/logger/td_logger.rb', line 70 def close unless @finish @finish = true @mutex.synchronize { @flush_now = true @cond.signal } @upload_thread.join if @upload_thread @queue.reject! {|db,table,data| begin upload(db, table, data) true rescue @logger.error "Failed to upload event logs to Treasure Data, trashed: #{$!}" false end } @map.reject! {|(db,table),buffer| data = buffer.flush! begin upload(db, table, data) true rescue @logger.error "Failed to upload event logs to Treasure Data, trashed: #{$!}" false end } end end |
#flush ⇒ Object
101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
# File 'lib/td/logger/td_logger.rb', line 101 def flush @mutex.lock # Move small chunks into queue to flush all events. # See try_flush routine for more detail @map.reject! {|(db,table),buffer| data = buffer.flush! @queue << [db, table, data] } try_flush rescue => e @logger.error "Unexpected error at flush: #{e}" e.backtrace.each {|bt| @logger.info bt } ensure @mutex.unlock end |
#post_with_time(tag, record, time) ⇒ Object
120 121 122 123 124 125 126 127 128 129 |
# File 'lib/td/logger/td_logger.rb', line 120 def post_with_time(tag, record, time) @logger.debug { "event: #{tag} #{record.to_json}" rescue nil } record[:time] ||= time.to_i tag = "#{@tag_prefix}.#{tag}" if @tag_prefix db, table = tag.split('.')[-2, 2] add(db, table, record) end |
#upload_main ⇒ Object
131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
# File 'lib/td/logger/td_logger.rb', line 131 def upload_main @mutex.lock until @finish now = Time.now.to_i if @next_time <= now || (@flush_now && @error_count == 0) flushed = try_flush @flush_now = false end if @error_count == 0 if flushed && @flush_interval < @max_flush_interval @flush_interval = [@flush_interval ** 2, @max_flush_interval].min end next_wait = @flush_interval else next_wait = @retry_wait * (2 ** (@error_count-1)) end @next_time = next_wait + now cond_wait(next_wait) end rescue @logger.error "Unexpected error: #{$!}" $!.backtrace.each {|bt| @logger.info bt } ensure @mutex.unlock end |