lib/mogilefs/mysql.rb

   1 require 'mogilefs'
   2 require 'mogilefs/backend' # for the exceptions
   3
   4 # read-only interface that can be a backend for MogileFS::MogileFS
   5 #
   6 # This provides direct, read-only access to any slave MySQL database to
   7 # provide better performance, scalability and eliminate mogilefsd as a
   8 # point of failure
   9 class MogileFS::Mysql
  10
  11   attr_reader :my
  12   attr_reader :query_method
  13
  14   ##
  15   # Creates a new MogileFS::Mysql instance.  +args+ must include a key
  16   # :domain specifying the domain of this client and :mysql, specifying
  17   # an already-initialized Mysql object.
  18   #
  19   # The Mysql object can be either the standard Mysql driver or the
  20   # Mysqlplus one supporting c_async_query.
  21   def initialize(args = {})
  22     @my = args[:mysql]
  23     @query_method = @my.respond_to?(:c_async_query) ? :c_async_query : :query
  24     @last_update_device = @last_update_domain = Time.at(0)
  25     @cache_domain = @cache_device = nil
  26   end
  27
  28   ##
  29   # Lists keys starting with +prefix+ follwing +after+ up to +limit+.  If
  30   # +after+ is nil the list starts at the beginning.
  31   def _list_keys(domain, prefix = '', after = '', limit = 1000, &block)
  32     # this code is based on server/lib/MogileFS/Worker/Query.pm
  33     dmid = refresh_domain[domain] or \
  34       raise MogileFS::Backend::DomainNotFoundError
  35
  36     # don't modify passed arguments
  37     limit ||= 1000
  38     limit = limit.to_i
  39     limit = 1000 if limit > 1000 || limit <= 0
  40     after, prefix = "#{after}", "#{prefix}"
  41
  42     if after.length > 0 && /^#{Regexp.quote(prefix)}/ !~ after
  43       raise MogileFS::Backend::AfterMismatchError
  44     end
  45
  46     raise MogileFS::Backend::InvalidCharsError if /[%\\]/ =~ prefix
  47     prefix.gsub!(/_/, '\_') # not sure why MogileFS::Worker::Query does this...
  48
  49     sql = <<-EOS
  50     SELECT dkey,length,devcount FROM file
  51     WHERE dmid = #{dmid}
  52       AND dkey LIKE '#{@my.quote(prefix)}%'
  53       AND dkey > '#{@my.quote(after)}'
  54     ORDER BY dkey LIMIT #{limit}
  55     EOS
  56
  57     keys = []
  58     query(sql).each do |dkey,length,devcount|
  59       yield(dkey, length, devcount) if block_given?
  60       keys << dkey
  61     end
  62
  63     keys.empty? ? nil : [ keys, (keys.last || '') ]
  64   end
  65
  66   ##
  67   # Returns the size of +key+.
  68   def _size(domain, key)
  69     dmid = refresh_domain[domain] or \
  70       raise MogileFS::Backend::DomainNotFoundError
  71
  72     sql = <<-EOS
  73     SELECT length FROM file
  74     WHERE dmid = #{dmid} AND dkey = '#{@my.quote(key)}'
  75     LIMIT 1
  76     EOS
  77
  78     res = query(sql).fetch_row
  79     return res[0].to_i if res && res[0]
  80     raise MogileFS::Backend::UnknownKeyError
  81   end
  82
  83   ##
  84   # Get the paths for +key+.
  85   def _get_paths(params = {})
  86     zone = params[:zone]
  87     noverify = (params[:noverify] == 1) # TODO this is unused atm
  88     dmid = refresh_domain[params[:domain]] or \
  89       raise MogileFS::Backend::DomainNotFoundError
  90     devices = refresh_device or raise MogileFS::Backend::NoDevicesError
  91     urls = []
  92     sql = <<-EOS
  93     SELECT fid FROM file
  94     WHERE dmid = #{dmid} AND dkey = '#{@my.quote(params[:key])}'
  95     LIMIT 1
  96     EOS
  97
  98     res = query(sql).fetch_row
  99     res && res[0] or raise MogileFS::Backend::UnknownKeyError
 100     fid = res[0]
 101     sql = "SELECT devid FROM file_on WHERE fid = '#{@my.quote(fid)}'"
 102     query(sql).each do |devid,|
 103       devinfo = devices[devid.to_i]
 104       port = devinfo[:http_get_port]
 105       host = zone && zone == 'alt' ? devinfo[:altip] : devinfo[:hostip]
 106       nfid = '%010u' % fid
 107       b, mmm, ttt = /(\d)(\d{3})(\d{3})(?:\d{3})/.match(nfid)[1..3]
 108       uri = "/dev#{devid}/#{b}/#{mmm}/#{ttt}/#{nfid}.fid"
 109       urls << "http://#{host}:#{port}#{uri}"
 110     end
 111     urls
 112   end
 113
 114   def sleep(params); Kernel.sleep(params[:duration] || 10); {}; end
 115
 116   private
 117
 118     unless defined? GET_DEVICES
 119       GET_DOMAINS = 'SELECT dmid,namespace FROM domain'.freeze
 120
 121       GET_DEVICES = <<-EOS
 122         SELECT d.devid, h.hostip, h.altip, h.http_port, h.http_get_port
 123         FROM device d
 124           LEFT JOIN host h ON d.hostid = h.hostid
 125         WHERE d.status IN ('alive','readonly','drain');
 126       EOS
 127       GET_DEVICES.freeze
 128     end
 129
 130     def query(sql)
 131       @my.send(@query_method, sql)
 132     end
 133
 134     def refresh_device(force = false)
 135       return @cache_device if ! force && ((Time.now - @last_update_device) < 60)
 136       tmp = {}
 137       res = query(GET_DEVICES)
 138       res.each do |devid, hostip, altip, http_port, http_get_port|
 139         http_port = http_port ? http_port.to_i : 80
 140         tmp[devid.to_i] = {
 141           :hostip => hostip.freeze,
 142           :altip => (altip || hostip).freeze,
 143           :http_port => http_port,
 144           :http_get_port => http_get_port ?  http_get_port.to_i : http_port,
 145         }.freeze
 146       end
 147       @last_update_device = Time.now
 148       @cache_device = tmp.freeze
 149     end
 150
 151     def refresh_domain(force = false)
 152       return @cache_domain if ! force && ((Time.now - @last_update_domain) < 5)
 153       tmp = {}
 154       res = query(GET_DOMAINS)
 155       res.each { |dmid,namespace| tmp[namespace] = dmid.to_i }
 156       @last_update_domain = Time.now
 157       @cache_domain = tmp.freeze
 158     end
 159
 160 end