(self, daemon_mode=False)
Copy partial WAL segments to slave.

        On 8.2 set use_xlog_functions=1 in config file - this enables record based
        walshipping. On 8.0 the only option is to sync files.

        If daemon_mode is specified it never switches from record based shipping to
        file based shipping.
Copy partial WAL segments to slave.
Copy partial WAL segments to slave .
def master_sync(self, daemon_mode=False):
        """
        Copy partial WAL segments to slave.

        On 8.2 set use_xlog_functions=1 in config file - this enables record based
        walshipping. On 8.0 the only option is to sync files.

        If daemon_mode is specified it never switches from record based shipping to
        file based shipping.
        """

        self.assert_is_master(True)

        use_xlog_functions = self.cf.getint("use_xlog_functions", False)
        data_dir = self.cf.getfile("master_data")
        xlog_dir = os.path.join(data_dir, "pg_xlog")
        master_bin = self.cf.getfile("master_bin", "")

        dst_loc = os.path.join(self.cf.getfile("partial_wals"), "")

        db = None
        if use_xlog_functions:
            try:
                db = self.get_database("master_db", autocommit=1)
            except:
                self.log.warning("Database unavailable, record based log shipping not possible.")
                if daemon_mode:
                    return

        if db:
            cur = db.cursor()
            cur.execute("select file_name, file_offset from pg_xlogfile_name_offset(pg_current_xlog_location())")
            (file_name, file_offs) = cur.fetchone()

            if not self.walchunk or self.walchunk.filename != file_name:
                # Switched to new WAL segment. Don't bother to copy the last bits - it
                # will be obsoleted by the archive_command.
                if self.walchunk and self.walchunk.sync_count > 0:
                    self.log.info("Switched in %d seconds, %f sec in %d interim syncs, avg %f",
                            time.time() - self.walchunk.start_time,
                            self.walchunk.sync_time,
                            self.walchunk.sync_count,
                            self.walchunk.sync_time / self.walchunk.sync_count)
                self.walchunk = WalChunk(file_name, 0, file_offs)
            else:
                self.walchunk.bytes = file_offs - self.walchunk.pos

            if self.walchunk.bytes > 0:
                self.master_send_partial(xlog_dir, self.walchunk, daemon_mode)
        else:
            files = os.listdir(xlog_dir)
            files.sort()

            last = self.get_last_complete()
            if last:
                self.log.info("%s: last complete", last)
            else:
                self.log.info("last complete not found, copying all")

            # obtain the last checkpoint wal name, this can be used for
            # limiting the amount of WAL files to copy if the database
            # has been cleanly shut down
            ctl = PgControlData(master_bin, data_dir, False)
            checkpoint_wal = None
            if ctl.is_valid:
                if not ctl.is_shutdown:
                    # cannot rely on the checkpoint wal, should use some other method
                    self.log.info("Database state is not 'shut down', copying all")
                else:
                    # ok, the database is shut down, we can use last checkpoint wal
                    checkpoint_wal = ctl.wal_name
                    self.log.info("last checkpoint wal: %s", checkpoint_wal)
            else:
                self.log.info("Unable to obtain control file information, copying all")

            for fn in files:
                # check if interesting file
                if len(fn) < 10:
                    continue
                if fn[0] < "0" or fn[0] > '9':
                    continue
                if fn.find(".") > 0:
                    continue
                # check if too old
                if last:
                    dot = last.find(".")
                    if dot > 0:
                        xlast = last[:dot]
                        if fn < xlast:
                            continue
                    else:
                        if fn <= last:
                            continue
                # check if too new
                if checkpoint_wal and fn > checkpoint_wal:
                    continue

                # got interesting WAL
                xlog = os.path.join(xlog_dir, fn)
                # copy data
                self.log.info('Syncing %s', xlog)
                if self.exec_rsync([xlog, dst_loc], not daemon_mode) != 0:
                    self.log.error('Cannot sync %s', xlog)
                    break
            else:
                self.log.info("Partial copy done")
def master_sync ( self , daemon_mode = False ) : self . assert_is_master ( True ) use_xlog_functions = self . cf . getint ( "use_xlog_functions" , False ) data_dir = self . cf . getfile ( "master_data" ) xlog_dir = os . path . join ( data_dir , "pg_xlog" ) master_bin = self . cf . getfile ( "master_bin" , "" ) dst_loc = os . path . join ( self . cf . getfile ( "partial_wals" ) , "" ) db = None if use_xlog_functions : try : db = self . get_database ( "master_db" , autocommit = 1 ) except : self . log . warning ( "Database unavailable, record based log shipping not possible." ) if daemon_mode : return if db : cur = db . cursor ( ) cur . execute ( "select file_name, file_offset from pg_xlogfile_name_offset(pg_current_xlog_location())" ) ( file_name , file_offs ) = cur . fetchone ( ) if not self . walchunk or self . walchunk . filename != file_name : # Switched to new WAL segment. Don't bother to copy the last bits - it # will be obsoleted by the archive_command. if self . walchunk and self . walchunk . sync_count > 0 : self . log . info ( "Switched in %d seconds, %f sec in %d interim syncs, avg %f" , time . time ( ) - self . walchunk . start_time , self . walchunk . sync_time , self . walchunk . sync_count , self . walchunk . sync_time / self . walchunk . sync_count ) self . walchunk = WalChunk ( file_name , 0 , file_offs ) else : self . walchunk . bytes = file_offs - self . walchunk . pos if self . walchunk . bytes > 0 : self . master_send_partial ( xlog_dir , self . walchunk , daemon_mode ) else : files = os . listdir ( xlog_dir ) files . sort ( ) last = self . get_last_complete ( ) if last : self . log . info ( "%s: last complete" , last ) else : self . log . info ( "last complete not found, copying all" ) # obtain the last checkpoint wal name, this can be used for # limiting the amount of WAL files to copy if the database # has been cleanly shut down ctl = PgControlData ( master_bin , data_dir , False ) checkpoint_wal = None if ctl . is_valid : if not ctl . is_shutdown : # cannot rely on the checkpoint wal, should use some other method self . log . info ( "Database state is not 'shut down', copying all" ) else : # ok, the database is shut down, we can use last checkpoint wal checkpoint_wal = ctl . wal_name self . log . info ( "last checkpoint wal: %s" , checkpoint_wal ) else : self . log . info ( "Unable to obtain control file information, copying all" ) for fn in files : # check if interesting file if len ( fn ) < 10 : continue if fn [ 0 ] < "0" or fn [ 0 ] > '9' : continue if fn . find ( "." ) > 0 : continue # check if too old if last : dot = last . find ( "." ) if dot > 0 : xlast = last [ : dot ] if fn < xlast : continue else : if fn <= last : continue # check if too new if checkpoint_wal and fn > checkpoint_wal : continue # got interesting WAL xlog = os . path . join ( xlog_dir , fn ) # copy data self . log . info ( 'Syncing %s' , xlog ) if self . exec_rsync ( [ xlog , dst_loc ] , not daemon_mode ) != 0 : self . log . error ( 'Cannot sync %s' , xlog ) break else : self . log . info ( "Partial copy done" )