lib/mogilefs/bigfile.rb

   1 require 'zlib'
   2 require 'digest/md5'
   3 require 'uri'
   4
   5 module MogileFS::Bigfile
   6   GZIP_HEADER = "\x1f\x8b".freeze # mogtool(1) has this
   7   # VALID_TYPES = %w(file tarball partition).map { |x| x.freeze }.freeze
   8
   9   def bigfile_stat(key)
  10     info = get_file_data(key)
  11     parse_info(info)
  12   end
  13
  14   # returns the big_info hash if successful, raises an exception if not
  15   def bigfile_write(key, wr, opts = { :verify => false })
  16     info = bigfile_stat(key)
  17     zi = nil
  18     md5 = opts[:verify] ? Digest::MD5.new : nil
  19     total = 0
  20
  21     # we only decode raw zlib deflated streams that mogtool (unfortunately)
  22     # generates.  tarballs and gzip(1) are up to to the application to decrypt.
  23     filter = Proc.new do |buf|
  24       if zi == nil
  25         if info[:compressed] && info[:type] == 'file' &&
  26              buf.length >= 2 && buf[0,2] != GZIP_HEADER
  27           zi = Zlib::Inflate.new
  28
  29           # mogtool(1) seems to have a bug that causes it to generate bogus
  30           # MD5s if zlib deflate is used.  Don't trust those MD5s for now...
  31           md5 = nil
  32         else
  33           zi = false
  34         end
  35       end
  36       buf ||= ''
  37       if zi
  38         zi.inflate(buf)
  39       else
  40         md5 << buf
  41         buf
  42       end
  43     end if (info[:compressed] || md5)
  44
  45     info[:parts].each_with_index do |part,part_nr|
  46       next if part_nr == 0 # info[:parts][0] is always empty
  47       uris = verify_uris(part[:paths].map { |path| URI.parse(path) })
  48       if uris.empty?
  49         # part[:paths] may not be valid anymore due to rebalancing, however we
  50         # can get_keys on key,<part_nr> and retry paths if all paths fail
  51         part[:paths] = get_paths("#{key.gsub(/^big_info:/, '')},#{part_nr}")
  52         uris = verify_uris(part[:paths].map { |path| URI.parse(path) })
  53         raise MogileFS::Backend::NoDevices if uris.empty?
  54       end
  55
  56       sock = http_get_sock(uris[0])
  57       md5.reset if md5
  58       w = sysrwloop(sock, wr, filter)
  59
  60       if md5 && md5.hexdigest != part[:md5]
  61         raise MogileFS::ChecksumMismatchError, "#{md5} != #{part[:md5]}"
  62       end
  63       total += w
  64     end
  65
  66     wr.syswrite(zi.finish) if zi
  67
  68     [ total, info ]
  69   end
  70
  71   private
  72
  73     def parse_info(info = '')
  74       rv = { :parts => [] }
  75       info.each_line do |line|
  76         line.chomp!
  77         case line
  78         when /^(des|type|filename)\s+(.+)$/
  79           rv[$1.to_sym] = $2
  80         when /^compressed\s+([01])$/
  81           rv[:compressed] = ($1 == '1')
  82         when /^(chunks|size)\s+(\d+)$/
  83           rv[$1.to_sym] = $2.to_i
  84         when /^part\s+(\d+)\s+bytes=(\d+)\s+md5=(.+)\s+paths:\s+(.+)$/
  85           rv[:parts][$1.to_i] = {
  86             :bytes => $2.to_i,
  87             :md5 => $3.downcase,
  88             :paths => $4.split(/\s*,\s*/),
  89           }
  90         end
  91       end
  92
  93       rv
  94     end
  95
  96 end # module MogileFS::Bigfile
  97
  98 __END__
  99 # Copied from mogtool:
 100 # http://code.sixapart.com/svn/mogilefs/utils/mogtool, r1221
 101
 102 # this is a temporary file that we delete when we're doing recording all chunks
 103
 104 _big_pre:<key>
 105
 106     starttime=UNIXTIMESTAMP
 107
 108 # when done, we write the _info file and delete the _pre.
 109
 110 _big_info:<key>
 111
 112     des Cow's ljdb backup as of 2004-11-17
 113     type  { partition, file, tarball }
 114     compressed {0, 1}
 115     filename  ljbinlog.305.gz
 116     partblocks  234324324324
 117
 118
 119     part 1 <bytes> <md5hex>
 120     part 2 <bytes> <md5hex>
 121     part 3 <bytes> <md5hex>
 122     part 4 <bytes> <md5hex>
 123     part 5 <bytes> <md5hex>
 124
 125 _big:<key>,<n>
 126 _big:<key>,<n>
 127 _big:<key>,<n>
 128
 129
 130 Receipt format:
 131
 132 BEGIN MOGTOOL RECIEPT
 133 type partition
 134 des Foo
 135 compressed foo
 136
 137 part 1 bytes=23423432 md5=2349823948239423984 paths: http://dev5/2/23/23/.fid, http://dev6/23/423/4/324.fid
 138 part 1 bytes=23423432 md5=2349823948239423984 paths: http://dev5/2/23/23/.fid, http://dev6/23/423/4/324.fid
 139 part 1 bytes=23423432 md5=2349823948239423984 paths: http://dev5/2/23/23/.fid, http://dev6/23/423/4/324.fid
 140 part 1 bytes=23423432 md5=2349823948239423984 paths: http://dev5/2/23/23/.fid, http://dev6/23/423/4/324.fid
 141
 142
 143 END RECIEPT
 144
 145
 146