def get_connection(self):
        """
        Connect to MySQL host and yield a connection.

        :return: MySQL connection
        :raise MySQLSourceError: if can't connect to server
        """
        connection = None
        try:
            connection = pymysql.connect(
                host=self.hostname,
                read_default_file=self.defaults_file,
                connect_timeout=self.connect_timeout,
                cursorclass=pymysql.cursors.DictCursor
            )

            yield connection
        except OperationalError:
            LOG.error(
                "Can't connect to MySQL server on %s",
                self.hostname)
            raise MySQLSourceError(
                "Can't connect to MySQL server on %s"
                % self.hostname)
        finally:
            if connection:
                connection.close()

def storage_server(docker_client, container_network):

    bootstrap_script = '/twindb-backup/support/bootstrap/storage_server.sh'
    container = get_container(
        'storage_server',
        docker_client,
        container_network,
        bootstrap_script=bootstrap_script,
        image="centos:centos7",
        last_n=2
    )

    timeout = time.time() + 30 * 60

    while time.time() < timeout:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        if sock.connect_ex((container['ip'], 22)) == 0:
            break
        time.sleep(1)

    yield container

    if container:
        LOG.info('Removing container %s', container['Id'])
        docker_client.api.remove_container(container=container['Id'],
                                           force=True)

    def disable_wsrep_desync(self):
        """
        Wait till wsrep_local_recv_queue is zero
        and disable wsrep_local_recv_queue then
        """
        max_time = time.time() + 900
        try:
            with self.get_connection() as connection:
                with connection.cursor() as cursor:
                    while time.time() < max_time:
                        cursor.execute("SHOW GLOBAL STATUS LIKE "
                                       "'wsrep_local_recv_queue'")

                        res = {r['Variable_name'].lower(): r['Value'].lower()
                               for r in cursor.fetchall()}

                        if not res.get('wsrep_local_recv_queue'):
                            raise Exception('Unknown status variable '
                                            '"wsrep_local_recv_queue"')

                        if int(res['wsrep_local_recv_queue']) == 0:
                            break

                        time.sleep(1)

                    LOG.debug('Disabling wsrep_desync')
                    cursor.execute("SET GLOBAL wsrep_desync=OFF")
        except pymysql.Error as err:
            LOG.error(err)

 def _handle_failure_exec(self, err, stderr_file):
     """Cleanup on failure exec"""
     LOG.error(err)
     LOG.error('Failed to run xtrabackup. '
               'Check error output in %s', stderr_file.name)
     self.dst.delete(self.get_name())
     exit(1)

    def get_stream(self):
        """
        Get a PIPE handler with content of the source
        :return:
        """
        cmd = [
            self._xtrabackup,
            "--defaults-file=%s" % self._connect_info.defaults_file,
            "--stream=xbstream",
            "--host=127.0.0.1",
            "--backup"
        ]
        cmd += ["--target-dir", "."]
        if self.is_galera():
            cmd.append("--galera-info")
            cmd.append("--no-backup-locks")
        if self.incremental:
            cmd += [
                "--incremental-basedir",
                ".",
                "--incremental-lsn=%d" % self._parent_lsn
            ]
        # If this is a Galera node then additional step needs to be taken to
        # prevent the backups from locking up the cluster.
        wsrep_desynced = False
        LOG.debug('Running %s', ' '.join(cmd))
        stderr_file = tempfile.NamedTemporaryFile(delete=False)
        try:
            if self.is_galera():
                wsrep_desynced = self.enable_wsrep_desync()

            LOG.debug('Running %s', ' '.join(cmd))
            proc_xtrabackup = Popen(cmd,
                                    stderr=stderr_file,
                                    stdout=PIPE)

            yield proc_xtrabackup.stdout

            proc_xtrabackup.communicate()
            if proc_xtrabackup.returncode:
                LOG.error('Failed to run xtrabackup. '
                          'Check error output in %s', stderr_file.name)
                try:
                    if LOG.debug_enabled:
                        with open(stderr_file.name) as xb_out:
                            for line in xb_out:
                                print(line, end='', file=sys.stderr)
                except AttributeError:
                    pass
                self.dst.delete(self.get_name())
                exit(1)
            else:
                LOG.debug('Successfully streamed xtrabackup output')
            self._update_backup_info(stderr_file)
        except OSError as err:
            LOG.error('Failed to run %s: %s', cmd, err)
            exit(1)
        finally:
            if wsrep_desynced:
                self.disable_wsrep_desync()

    def setup_slave(self, master_info):  # noqa # pylint: disable=too-many-arguments
        """
        Change master

        :param master_info: Master details.
        :type master_info: MySQLMasterInfo

        """
        try:
            with self._cursor() as cursor:
                query = "CHANGE MASTER TO " \
                        "MASTER_HOST = '{master}', " \
                        "MASTER_USER = '{user}', " \
                        "MASTER_PORT = {port}, " \
                        "MASTER_PASSWORD = '{password}', " \
                        "MASTER_LOG_FILE = '{binlog}', " \
                        "MASTER_LOG_POS = {binlog_pos}"\
                    .format(
                        master=master_info.host,
                        user=master_info.user,
                        password=master_info.password,
                        binlog=master_info.binlog,
                        binlog_pos=master_info.binlog_position,
                        port=master_info.port
                    )
                cursor.execute(query)
                cursor.execute("START SLAVE")
            return True
        except pymysql.Error as err:
            LOG.debug(err)
            return False

    def apply_backup(self, datadir):
        """
        Apply backup of destination server

        :param datadir: Path to datadir
        :return: Binlog file name and position
        :rtype: tuple
        :raise RemoteMySQLSourceError: if any error.
        """
        try:
            use_memory = "--use-memory %d" % int(self._mem_available() / 2)
        except OSError:
            use_memory = ""
        logfile_path = "/tmp/xtrabackup-apply-log.log"
        cmd = "sudo {xtrabackup} --prepare --apply-log-only " \
              "--target-dir {target_dir} {use_memory} " \
              "> {logfile} 2>&1" \
              "".format(
                  xtrabackup=self._xtrabackup,
                  target_dir=datadir,
                  use_memory=use_memory,
                  logfile=logfile_path
              )

        try:
            self._ssh_client.execute(cmd)
            self._ssh_client.execute("sudo chown -R mysql %s" % datadir)
            return self._get_binlog_info(datadir)
        except SshClientException as err:
            LOG.debug("Logfile is:")
            LOG.debug(self._ssh_client.get_text_content(logfile_path))
            raise RemoteMySQLSourceError(err)

def backup_files(run_type, config):
    """Backup local directories

    :param run_type: Run type
    :type run_type: str
    :param config: Configuration
    :type config: TwinDBBackupConfig
    """
    backup_start = time.time()
    try:
        for directory in config.backup_dirs:
            LOG.debug('copying %s', directory)
            src = FileSource(directory, run_type)
            dst = config.destination()
            _backup_stream(config, src, dst)
            src.apply_retention_policy(dst, config, run_type)
    except (
            DestinationError,
            SourceError,
            SshClientException
    ) as err:
        raise OperationError(err)
    export_info(config, data=time.time() - backup_start,
                category=ExportCategory.files,
                measure_type=ExportMeasureType.backup)

def backup_everything(run_type, twindb_config, binlogs_only=False):
    """
    Run backup job

    :param run_type: hourly, daily, etc
    :type run_type: str
    :param twindb_config: ConfigParser instance
    :type twindb_config: TwinDBBackupConfig
    :param binlogs_only: If True copy only MySQL binary logs.
    :type binlogs_only: bool
    """
    set_open_files_limit()

    try:
        if not binlogs_only:
            backup_start = time.time()
            backup_files(run_type, twindb_config)
            backup_mysql(run_type, twindb_config)
            backup_binlogs(run_type, twindb_config)
            end = time.time()
            save_measures(backup_start, end)
        else:
            backup_binlogs(run_type, twindb_config)
    except ConfigParser.NoSectionError as err:
        LOG.debug(traceback.format_exc())
        LOG.error(err)
        exit(1)

 def backup_mysql(self):
     """FLag to backup MySQL or not"""
     try:
         return self.__cfg.getboolean('source', 'backup_mysql')
     except NoOptionError:
         return False
     except NoSectionError as err:
         LOG.error("Section 'source' is mandatory")
         raise ConfigurationError(err)

    def _update_backup_info(self, stderr_file):
        """Update backup_info from stderr"""

        LOG.debug('xtrabackup error log file %s',
                  stderr_file.name)
        self._backup_info.lsn = self._get_lsn(stderr_file.name)
        self._backup_info.binlog_coordinate = self.get_binlog_coordinates(
            stderr_file.name
        )
        os.unlink(stderr_file.name)

    def clone_config(self, dst):
        """
        Clone config to destination server

        :param dst: Destination server
        :type dst: Ssh
        """
        cfg_path = self._get_root_my_cnf()
        LOG.debug("Root my.cnf is: %s", cfg_path)
        self._save_cfg(dst, cfg_path)

    def save(self, handler, filepath):
        """
        Read from handler and save it to Amazon S3

        :param filepath: save backup copy in a file with this name
        :param handler: stdout handler from backup source
        """
        with handler as file_obj:
            ret = self._upload_object(file_obj, filepath)
            LOG.debug('Returning code %d', ret)

def set_open_files_limit():
    """Detect maximum supported number of open file and set it"""
    max_files = getrlimit(RLIMIT_NOFILE)[0]
    while True:
        try:
            setrlimit(RLIMIT_NOFILE, (max_files, max_files))
            max_files += 1
        except ValueError:
            break
    LOG.debug('Setting max files limit to %d', max_files)

 def backup_dirs(self):
     """Directories to backup"""
     try:
         dirs = self.__cfg.get('source', 'backup_dirs')
         return split(dirs)
     except NoOptionError:
         return []
     except NoSectionError as err:
         LOG.error("Section 'source' is mandatory")
         raise ConfigurationError(err)

def kill_children():
    """
    Kill child process
    """
    for proc in multiprocessing.active_children():
        LOG.info('Terminating %r [%d] ...', proc, proc.pid)
        proc.terminate()
    parent = psutil.Process(os.getpid())
    for child in parent.children(recursive=True):
        LOG.info('Terminating process %r', child)
        child.kill()

def _print_binlog(dst):
    dst_files = dst.list_files(
        dst.remote_path,
        pattern='/binlog/',
        recursive=True,
        files_only=True
    )
    if dst_files:
        LOG.info('Binary logs:')
        for copy in dst_files:
            print(copy)

def share_backup(ctx, s3_url):
    """Share backup copy for download"""
    if not s3_url:
        LOG.info('No backup copy specified. Choose one from below:')
        list_available_backups(ctx.obj['twindb_config'])
        exit(1)
    try:
        share(ctx.obj['twindb_config'], s3_url)
    except TwinDBBackupError as err:
        LOG.error(err)
        exit(1)

def bucket_name():
    travis_job_number = os.environ.get('TRAVIS_JOB_NUMBER')
    LOG.debug('TRAVIS_JOB_NUMBER=%s' % travis_job_number)

    number = random.randint(0, 1000000)
    LOG.debug('Default job number %d' % number)

    if travis_job_number:
        bucket = 'twindb-backup-test-travis-%s' % travis_job_number
    else:
        bucket = 'twindb-backup-test-travis-%d' % number

    return '%s-%s' % (bucket, time.time())

def _print_media_type(dst, media_type):
    for run_type in INTERVALS:
        pattern = "/%s/%s/" % (run_type, media_type)
        dst_files = dst.list_files(
            dst.remote_path,
            pattern=pattern,
            recursive=True,
            files_only=True
        )
        if dst_files:
            LOG.info('%s %s copies:', media_type, run_type)
            for copy in dst_files:
                print(copy)