mysql backups

agent.rb

%w[sequel aws-sdk socket logger json pathname config upload].each {|l| require l}

module MysqlBackup
  class Agent
    include Config

    attr_accessor :summary, :pool, :path, :log, :db

    def initialize pool
      # intent
      @pool = pool
      @path = File.join config[:backup][:tmp], "mysql_backup_#{date}.tgz"
      @summary = { checkpoint: Time.now }

      # setup resources
      record :setup do
        [:log, :db, :disk, :network].each do |resource|
          setup resource rescue error "unable to setup resource: #{resource}"
        end
      end

      # sanity checks
      error 'mysql is not running, aborting backup' unless mysql :status
      error 'node is currently in use by app, aborting backup' if node_in_use?
    end

    def date
      @date ||= Time.now.strftime config[:backup][:time_format]
    end

    def error message
      log.error message
      abort message
    end

    def record event
      summary[:timers] ||= Hash.new 0
      checkpoint = Time.now

      yield
      summary[:timers][event] += (Time.now - checkpoint)
    end

    def summarize
      # record total time
      summary[:timers][:total] = Time.now - summary[:checkpoint]

      # merge in other meta data
      summary.merge!({
        pool: pool,
        host: Socket.gethostname,
      })

      # record our summary
      File.open(File.join(config[:backup][:tmp], config[:backup][:summary]), 'w') {|f| f.write summary.to_json}

      puts summary.to_json
    end

    def setup resource
      case resource
      when :log
        @log = Logger.new config[:log][:file], config[:log][:retention]
        @log.level = config[:log][:level]
      when :db
        log.info 'setting up mysql connection'

        @db = Sequel.connect config[:mysql]
      when :disk
        log.info 'setting disk scheduler'

        scheduler = File.join '/sys/block', config[:disk][:device], 'queue/scheduler'
        File.open scheduler, 'w' do |file|
          file.write config[:disk][:scheduler]
        end
      when :network
        log.info 'enabling public network interface'

        output, status = %x[ifup #{config[:network][:device]}], $?
        status.success?
      end
    end

    def cleanup resource
      case resource
      when :network
        log.info 'disabling public network interface'

        output, status = %x[ifdown #{config[:network][:device]}], $?
        status.success?
      when :files
        log.info 'deleting local copys of backups'

        output, status = %x[rm -f #{config[:backup][:tmp]}/mysql_backup*.tgz], $?
        status.success?
      end
    end

    def mysql action
      case action
      when :start
        log.info 'starting mysqld'

        record :mysql do
          output, status = %x[/etc/init.d/mysql start], $?
          error 'mysqld failed to start' unless status.success?
        end
      when :stop
        log.info 'stopping mysqld'

        record :mysql do
          output, status = %x[mysqladmin shutdown], $?
          error 'mysqld failed to shutdown' unless status.success?
        end
      when :status
        output, status = %x[mysqladmin status], $?
        status.success?
      end
    end

    def node_in_use?
      log.info 'querying process list to ensure node is not in use'

      db['show processlist'].any? do |process|
        process[:User] == config[:mysql][:app_user]
      end
    end

    def package
      log.info 'packaging mysqld dataset for backup'

      command = [
        "tar  --create #{config[:backup][:location]} 2> /dev/null",
        "pigz --stdout > #{path}",
      ].join ' | '

      record :package do
        output, status = %x[#{command}], $?
        status.success?
      end
    end

    def upload
      log.info 'starting upload to s3'

      record :upload do
        error 'upload failed' unless Upload.new pool, path
      end

      log.info 'upload complete'
    end

    def backup
      log.info 'starting mysqld backup process'

      # package up the dataset
      mysql :stop
      package
      mysql :start

      # ship to s3
      upload

      # clean up our and normalize
      record :clean_up do
        [:network, :files].each do |resource|
          cleanup resource rescue error "unable to cleanup resource: #{resource}"
        end
      end

      log.info 'mysqld backup complete'
      summarize
    end

  end
end

config.rb

module MysqlBackup
  module Config

    def config
      @config ||= {

        # log settings
        log: {
          file:      '/var/log/backup_agent.log',
          retention: 'weekly',
          level:     Logger::INFO,
        },

        # s3 settings
        s3: {
          bucket: 'xxx',
          access_key: 'xxx',
          secret_key: 'xxx',
        },

        # mysql settings
        mysql: {
          adapter: 'mysql2',
          host: 'localhost',
          user: 'root',
          password: 'xxx',

          app_user: 'xxx',
        },

        # disk settings
        disk: {
          device:    'sda',
          scheduler: 'cfq',
        },

        network: {
          device:    'bond1',
        },

        # backup settings
        backup: {
          location:    '/var/lib/mysql',
          summary:     'mysql_backup_summary.json',
          tmp:         '/var/tmp',
          threads:     40,
          chunk_size:  50 * 1024 * 1024, # MB
          time_format: '%Y-%m-%d_%H-%M', # 2012-05-01_14-45
        },

      }
    end

  end
end

mysql_backup.rb

# mysql backup agent
$:.unshift File.join File.dirname(__FILE__), 'mysql_backup'
require 'agent'

upload.rb

module MysqlBackup
  class Upload
    include Config

    attr_accessor :s3, :bucket, :object, :file

    def initialize pool, path
      AWS.config :access_key_id     => config[:s3][:access_key],
                 :secret_access_key => config[:s3][:secret_key]

      @file   = Pathname.new path
      @key    = File.join pool, @file.basename 

      @s3     = AWS::S3.new
      @bucket = @s3.buckets[config[:s3][:bucket]]
      @object = @bucket.objects[@key]

      enqueue && process
    end

    def queue
      @queue ||= Queue.new
    end

    def enqueue
      (file.size.to_f / config[:backup][:chunk_size]).ceil.times do |index|
        queue << [config[:backup][:chunk_size] * index, index + 1]
      end
    end

    def process
      object.multipart_upload do |upload|
        threads = []

        config[:backup][:threads].times do
          threads << Thread.new do
            until queue.empty?
              offset, index = queue.deq :asynchronously rescue nil

              unless offset.nil?
                upload.add_part :data => file.read(config[:backup][:chunk_size], offset),
                                :part_number => index
              end
            end
          end
        end

        threads.each &:join
      end
    end

  end
end