Class: TreasureData::Logger::TreasureDataLogger

Inherits:
Fluent::Logger::LoggerBase
  • Object
show all
Extended by:
Finalizable
Defined in:
lib/td/logger/tdlog.rb

Defined Under Namespace

Modules: Finalizable

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Finalizable

finalizer, new

Constructor Details

#initialize(apikey, tag, auto_create_table) ⇒ TreasureDataLogger

Returns a new instance of TreasureDataLogger.



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
 
# File 'lib/td/logger/tdlog.rb', line 26

def initialize(apikey, tag, auto_create_table)
  require 'thread'
  require 'stringio'
  require 'zlib'
  require 'msgpack'
  require 'time'
  require 'net/http'
  require 'cgi'
  require 'logger'
  require 'td/client'

  @tag = tag
  @auto_create_table = auto_create_table

  @logger = ::Logger.new(STDERR)
  @logger.level = ::Logger::INFO

  @client = TreasureData::Client.new(apikey)

  @mutex = Mutex.new
  @cond = ConditionVariable.new
  @map = {}  # (db,table) => buffer:String
  @queue = []

  @chunk_limit = 8*1024*1024
  @flush_interval = 10
  @max_flush_interval = 300
  @retry_wait = 1.0
  @retry_limit = 8

  @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
end

Instance Attribute Details

#logger ⇒ Object

Returns the value of attribute logger.



65
66
67
 
# File 'lib/td/logger/tdlog.rb', line 65

def logger
  @logger
end

Instance Method Details

#close ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/td/logger/tdlog.rb', line 67

def close
  unless @finish
    @finish = true
    @mutex.synchronize {
      @flush_now = true
      @cond.signal
    }
    @upload_thread.join if @upload_thread

    @map.each {|(db,table),buffer|
      upload(db, table, buffer)
    }
    @queue.each {|tuple|
      upload(*tuple)
    }
  end
end

#post(tag, record) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/td/logger/tdlog.rb', line 85

def post(tag, record)
  record[:time] ||= Time.now.to_i

  tag = "#{@tag}.#{tag}"
  db, table = tag.split('.')[-2, 2]

  key = [db, table]
  @mutex.synchronize do
    buffer = (@map[key] ||= '')
    record.to_msgpack(buffer)

    if buffer.size > @chunk_limit
      @queue << [db, table, buffer]
      @map.delete(key)
      @cond.signal
    end

    # stat upload thread if it's not run
    unless @upload_thread
      @upload_thread = Thread.new(&method(:upload_main))
    end
  end

  nil
end

#upload_main ⇒ Object 



111
112
113
114
115
116
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
143
144
145
146
 
# File 'lib/td/logger/tdlog.rb', line 111

def upload_main
  @mutex.lock
  until @finish
    now = Time.now.to_i

    if @next_time <= now || (@flush_now && @error_count == 0)
      @mutex.unlock
      begin
        flushed = try_flush
      ensure
        @mutex.lock
      end
      @flush_now = false
    end

    if @error_count == 0
      if flushed && @flush_interval < @max_flush_interval
        @flush_interval = [@flush_interval + 60, @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