pipeline: make errors easier to debug by including the request
[ruby-mogilefs-client.git] / lib / mogilefs / backend.rb
blob89f88a3679c0de74052576751f188d0d289b2158
1 # -*- encoding: binary -*-
2 require 'thread'
4 # This class communicates with the MogileFS trackers.
5 # You should not have to use this directly unless you are developing
6 # support for new commands or plugins for MogileFS
7 class MogileFS::Backend
9   # Adds MogileFS commands +names+.
10   def self.add_command(*names)
11     names.each do |name|
12       define_method name do |*args|
13         do_request(name, args[0] || {}, false)
14       end
15     end
16   end
18   # adds idempotent MogileFS commands +names+, these commands may be retried
19   # transparently on a different tracker if there is a network/server error.
20   def self.add_idempotent_command(*names)
21     names.each do |name|
22       define_method name do |*args|
23         do_request(name, args[0] || {}, true)
24       end
25     end
26   end
28   BACKEND_ERRORS = {} # :nodoc:
30   # this converts an error code from a mogilefsd tracker to an exception:
31   #
32   # Examples of some exceptions that get created:
33   #   class AfterMismatchError < MogileFS::Error; end
34   #   class DomainNotFoundError < MogileFS::Error; end
35   #   class InvalidCharsError < MogileFS::Error; end
36   def self.add_error(err_snake)
37     err_camel = err_snake.gsub(/(?:^|_)([a-z])/) { $1.upcase }
38     err_camel << 'Error' unless /Error\z/ =~ err_camel
39     unless const_defined?(err_camel)
40       const_set(err_camel, Class.new(MogileFS::Error))
41     end
42     BACKEND_ERRORS[err_snake] = const_get(err_camel)
43   end
45   ##
46   # The last error
48   attr_reader :lasterr
50   ##
51   # The string attached to the last error
53   attr_reader :lasterrstr
55   ##
56   # Creates a new MogileFS::Backend.
57   #
58   # :hosts is a required argument and must be an Array containing one or more
59   # 'hostname:port' pairs as Strings.
60   #
61   # :timeout adjusts the request timeout before an error is returned.
63   def initialize(args)
64     @hosts = args[:hosts]
65     raise ArgumentError, "must specify at least one host" unless @hosts
66     raise ArgumentError, "must specify at least one host" if @hosts.empty?
67     unless @hosts == @hosts.select { |h| h =~ /:\d+$/ } then
68       raise ArgumentError, ":hosts must be in 'host:port' form"
69     end
71     @mutex = Mutex.new
72     @timeout = args[:timeout] || 3
73     @socket = nil
74     @lasterr = nil
75     @lasterrstr = nil
76     @pending = []
78     @dead = {}
79   end
81   ##
82   # Closes this backend's socket.
84   def shutdown
85     @mutex.synchronize { shutdown_unlocked }
86   end
88   # MogileFS::MogileFS commands
90   add_command :create_open
91   add_command :create_close
92   add_idempotent_command :get_paths
93   add_command :delete
94   add_idempotent_command :sleep
95   add_command :rename
96   add_idempotent_command :list_keys
97   add_idempotent_command :file_info
98   add_idempotent_command :file_debug
100   # MogileFS::Backend commands
102   add_idempotent_command :get_hosts
103   add_idempotent_command :get_devices
104   add_idempotent_command :list_fids
105   add_idempotent_command :stats
106   add_idempotent_command :get_domains
107   add_command :create_domain
108   add_command :delete_domain
109   add_command :create_class
110   add_command :update_class
111   add_command :delete_class
112   add_command :create_host
113   add_command :update_host
114   add_command :delete_host
115   add_command :set_state
117   # Errors copied from MogileFS/Worker/Query.pm
118   add_error 'dup'
119   add_error 'after_mismatch'
120   add_error 'bad_params'
121   add_error 'class_exists'
122   add_error 'class_has_files'
123   add_error 'class_not_found'
124   add_error 'db'
125   add_error 'domain_has_files'
126   add_error 'domain_exists'
127   add_error 'domain_not_empty'
128   add_error 'domain_not_found'
129   add_error 'failure'
130   add_error 'host_exists'
131   add_error 'host_mismatch'
132   add_error 'host_not_empty'
133   add_error 'host_not_found'
134   add_error 'invalid_chars'
135   add_error 'invalid_checker_level'
136   add_error 'invalid_mindevcount'
137   add_error 'key_exists'
138   add_error 'no_class'
139   add_error 'no_devices'
140   add_error 'no_domain'
141   add_error 'no_host'
142   add_error 'no_ip'
143   add_error 'no_key'
144   add_error 'no_port'
145   add_error 'none_match'
146   add_error 'plugin_aborted'
147   add_error 'state_too_high'
148   add_error 'size_verify_error'
149   add_error 'unknown_command'
150   add_error 'unknown_host'
151   add_error 'unknown_key'
152   add_error 'unknown_state'
153   add_error 'unreg_domain'
155   def shutdown_unlocked(do_raise = false) # :nodoc:
156     @pending = []
157     if @socket
158       @socket.close rescue nil # ignore errors
159       @socket = nil
160     end
161     raise if do_raise
162   end
164   def dispatch_unlocked(request, timeout = @timeout) # :nodoc:
165     begin
166       io = socket
167       io.timed_write(request, timeout)
168       io
169     rescue SystemCallError, MogileFS::RequestTruncatedError  => err
170       @dead[@active_host] = [ Time.now, err ]
171       shutdown_unlocked
172       retry
173     end
174   end
176   def pipeline_gets_unlocked(io, timeout) # :nodoc:
177     line = io.timed_gets(timeout) or
178       raise MogileFS::PipelineError,
179             "EOF with #{@pending.size} requests in-flight"
180     ready = @pending.shift
181     ready[1].call(parse_response(line, ready[0]))
182   end
184   def timeout_update(timeout, t0) # :nodoc:
185     timeout -= (Time.now - t0)
186     timeout < 0 ? 0 : timeout
187   end
189   # try to read any responses we have pending already before filling
190   # the pipeline more requests.  This usually takes very little time,
191   # but trackers may return huge responses and we could be on a slow
192   # network.
193   def pipeline_drain_unlocked(io, timeout) # :nodoc:
194     set = [ io ]
195     while @pending.size > 0
196       t0 = Time.now
197       r = IO.select(set, set, nil, timeout)
198       timeout = timeout_update(timeout, t0)
200       if r && r[0][0]
201         t0 = Time.now
202         pipeline_gets_unlocked(io, timeout)
203         timeout = timeout_update(timeout, t0)
204       else
205         return timeout
206       end
207     end
208     timeout
209   end
211   # dispatch a request like do_request, but queue +block+ for execution
212   # upon receiving a response.
213   def pipeline_dispatch(cmd, args, &block) # :nodoc:
214     request = make_request(cmd, args)
215     timeout = @timeout
217     @mutex.synchronize do
218       io = socket
219       timeout = pipeline_drain_unlocked(io, timeout)
221       # send the request out...
222       begin
223         io.timed_write(request, timeout)
224         @pending << [ request, block ]
225       rescue SystemCallError, MogileFS::RequestTruncatedError => err
226         @dead[@active_host] = [ Time.now, err ]
227         shutdown_unlocked(@pending[0])
228         io = socket
229         retry
230       end
232       @pending.size
233     end
234   end
236   def pipeline_wait(count = nil) # :nodoc:
237     @mutex.synchronize do
238       io = socket
239       count ||= @pending.size
240       @pending.size < count and
241         raise MogileFS::PipelineError,
242               "pending=#{@pending.size} < expected=#{count} failed"
243       begin
244         count.times { pipeline_gets_unlocked(io, @timeout) }
245       rescue
246         shutdown_unlocked(true)
247       end
248     end
249   end
251   # Performs the +cmd+ request with +args+.
252   def do_request(cmd, args, idempotent = false)
253     request = make_request cmd, args
254     @mutex.synchronize do
255       begin
256         io = dispatch_unlocked(request)
257         line = io.timed_gets(@timeout) and return parse_response(line)
259         idempotent or
260           raise EOFError, "end of file reached after: #{request.inspect}"
261         # fall through to retry in loop
262       rescue SystemCallError,
263              MogileFS::UnreadableSocketError,
264              MogileFS::InvalidResponseError, # truncated response
265              MogileFS::Timeout
266         # we got a successful timed_write, but not a timed_gets
267         retry if idempotent
268         shutdown_unlocked(true)
269       rescue
270         # we DO NOT want the response we timed out waiting for, to crop up later
271         # on, on the same socket, intersperesed with a subsequent request!  we
272         # close the socket if there's any error.
273         shutdown_unlocked(true)
274       end while idempotent
275     end # @mutex.synchronize
276   end
278   # Makes a new request string for +cmd+ and +args+.
279   def make_request(cmd, args)
280     "#{cmd} #{url_encode args}\r\n"
281   end
283   # this converts an error code from a mogilefsd tracker to an exception
284   # Most of these exceptions should already be defined, but since the
285   # MogileFS server code is liable to change and we may not always be
286   # able to keep up with the changes
287   def error(err_snake)
288     BACKEND_ERRORS[err_snake] || self.class.add_error(err_snake)
289   end
291   # Turns the +line+ response from the server into a Hash of options, an
292   # error, or raises, as appropriate.
293   def parse_response(line, request = nil)
294     if line =~ /^ERR\s+(\w+)\s*([^\r\n]*)/
295       @lasterr = $1
296       @lasterrstr = $2 ? url_unescape($2) : nil
297       if request
298         request = " request=#{request.strip}"
299         @lasterrstr = @lasterrstr ? (@lasterrstr << request) : request
300         return error(@lasterr).new(@lasterrstr)
301       end
302       raise error(@lasterr).new(@lasterrstr)
303     end
305     return url_decode($1) if line =~ /^OK\s+\d*\s*(\S*)\r\n\z/
307     raise MogileFS::InvalidResponseError,
308           "Invalid response from server: #{line.inspect}"
309   end
311   # Returns a socket connected to a MogileFS tracker.
312   def socket
313     return @socket if @socket and not @socket.closed?
315     now = Time.now
317     @hosts.shuffle.each do |host|
318       next if @dead.include?(host) and @dead[host][0] > now - 5
320       begin
321         addr, port = host.split(/:/)
322         @socket = MogileFS::Socket.tcp(addr, port, @timeout)
323         @active_host = host
324       rescue SystemCallError, MogileFS::Timeout => err
325         @dead[host] = [ now, err ]
326         next
327       end
329       return @socket
330     end
332     errors = @dead.map { |host,(_,e)| "#{host} - #{e.message} (#{e.class})" }
333     raise MogileFS::UnreachableBackendError,
334           "couldn't connect to any tracker: #{errors.join(', ')}"
335   end
337   # Turns a url params string into a Hash.
338   def url_decode(str) # :nodoc:
339     Hash[*(str.split(/&/).map! { |pair|
340       pair.split(/=/, 2).map! { |x| url_unescape(x) }
341     } ).flatten]
342   end
344   # :stopdoc:
345   # TODO: see if we can use existing URL-escape/unescaping routines
346   # in the Ruby standard library, Perl MogileFS seems to NIH these
347   #  routines, too
348   # :startdoc:
350   # Turns a Hash (or Array of pairs) into a url params string.
351   def url_encode(params) # :nodoc:
352     params.map do |k,v|
353       "#{url_escape k.to_s}=#{url_escape v.to_s}"
354     end.join("&")
355   end
357   # Escapes naughty URL characters.
358   if ''.respond_to?(:ord) # Ruby 1.9
359     def url_escape(str) # :nodoc:
360       str.gsub(/([^\w\,\-.\/\\\: ])/) { "%%%02x" % $1.ord }.tr(' ', '+')
361     end
362   else # Ruby 1.8
363     def url_escape(str) # :nodoc:
364       str.gsub(/([^\w\,\-.\/\\\: ])/) { "%%%02x" % $1[0] }.tr(' ', '+')
365     end
366   end
368   # Unescapes naughty URL characters.
369   def url_unescape(str) # :nodoc:
370     str.tr('+', ' ').gsub(/%([a-f0-9][a-f0-9])/i) { [$1.to_i(16)].pack 'C' }
371   end