management/python/bin/qpid-store-resize

#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

from qpidstore import jerr, jrnl, janal
import glob, optparse, os, sys, time


#== class Resize ==============================================================

class Resize(object):
    """
    Creates a new store journal and copies records from old journal to new. The new journal may be of
    different size from the old one. The records are packed into the new journal (ie only remaining
    enqueued records and associated transactions - if any - are copied over without spaces between them).
    
    The default action is to push the old journal down into a 'bak' sub-directory and then create a
    new journal of the same size and pack it with the records from the old. However, it is possible to
    suppress the pushdown (using --no-pushdown), in which case either a new journal id (using
    --new-base-filename) or an old journal id (usnig --old-base-filename) must be supplied. In the former
    case,a new journal will be created using the new base file name alongside the old one. In the latter
    case, the old journal will be renamed to the supplied name, and the new one will take the default.
    Note that both can be specified together with the --no-pushdown option.
    
    To resize the journal, use the optional --num-jfiles and/or --jfile-size parameters. These
    should be large enough to write all the records or an error will result. If the size is large enough
    to write all records, but too small to keep below the enqueue threshold, a warning will be printed.
    Note that as any valid size will be accepted, a journal can also be shrunk, as long as it is sufficiently
    big to accept the transferred records.
    """
    
    BAK_DIR = "bak"
    JFILE_SIZE_PGS_MIN = 1
    JFILE_SIZE_PGS_MAX = 32768
    NUM_JFILES_MIN = 4
    NUM_JFILES_MAX = 64
    
    def __init__(self):
        """Constructor"""
        self._opts = None
        self._jdir = None
        self._fname = None
        self._fnum = None
        self._file = None
        self._file_rec_wr_cnt = None
        self._filler_wr_cnt = None
        self._last_rec_fid = None
        self._last_rec_offs = None
        self._rec_wr_cnt = None
        
        self._jrnl_info = None
        self._jrnl_analysis = None
        self._jrnl_reader = None
        
        self._process_args()
        self._jrnl_info = jrnl.JrnlInfo(self._jdir, self._opts.bfn)
        # FIXME: This is a hack... find an elegant way of getting the file size to jrec!
        jrnl.JRNL_FILE_SIZE = self._jrnl_info.get_jrnl_file_size_bytes()
        self._jrnl_analysis = janal.JrnlAnalyzer(self._jrnl_info)
        self._jrnl_reader = janal.JrnlReader(self._jrnl_info, self._jrnl_analysis, self._opts.qflag, self._opts.rflag,
                                            self._opts.vflag)
    
    def run(self):
        """Perform the action of resizing the journal"""
        if not self._opts.qflag:
            print self._jrnl_analysis
        self._jrnl_reader.run()
        if self._opts.vflag:
            print self._jrnl_info
        if not self._opts.qflag:
            print self._jrnl_reader.report(self._opts.vflag, self._opts.rflag)
        self._handle_old_files()
        self._create_new_files()
        if not self._opts.qflag:
            print "Transferred %d records to new journal." % self._rec_wr_cnt
        self._chk_free()
    
    def _chk_free(self):
        """Check if sufficient space is available in resized journal to be able to enqueue. Raise a warning if not."""
        if self._last_rec_fid == None or self._last_rec_offs == None:
            return
        wr_capacity_bytes = self._last_rec_fid * self._jrnl_info.get_jrnl_data_size_bytes() + self._last_rec_offs
        tot_capacity_bytes = self._jrnl_info.get_tot_jrnl_data_size_bytes()
        percent_full = 100.0 * wr_capacity_bytes / tot_capacity_bytes
        if percent_full > 80.0:
            raise jerr.JWarning("WARNING: Journal %s is %2.1f%% full and will likely not allow enqueuing of new records"
                                " until some existing records are dequeued." %
                                (self._jrnl_info.get_jrnl_id(), percent_full))
    
    def _create_new_files(self):
        """Create new journal files"""
        # Assemble records to be transferred
        master_record_list = {}
        txn_record_list = self._jrnl_reader.txn_obj_list()
        if self._opts.vflag and self._jrnl_reader.emap().size() > 0:
            print "* Assembling %d records from emap" % self._jrnl_reader.emap().size()
        for tup in self._jrnl_reader.emap().get_rec_list():
            hdr = tup[1]
            hdr.flags &= ~jrnl.Hdr.OWI_MASK # Turn off owi
            master_record_list[long(hdr.rid)] = hdr
            if hdr.xidsize > 0 and hdr.xid in txn_record_list:
                txn_hdr = txn_record_list[hdr.xid]
                del(txn_record_list[hdr.xid])
                txn_hdr.flags &= ~jrnl.Hdr.OWI_MASK # Turn off owi
                master_record_list[long(txn_hdr.rid)] = txn_hdr
        if self._opts.vflag and self._jrnl_reader.tmap().size() > 0:
            print "* Assembling %d records from tmap" % self._jrnl_reader.tmap().size()
        for xid in self._jrnl_reader.tmap().xids():
            for l in self._jrnl_reader.tmap().get(xid):
                hdr = l[1]
                hdr.flags &= ~jrnl.Hdr.OWI_MASK # Turn off owi
                master_record_list[hdr.rid] = hdr
        rid_list = master_record_list.keys()
        rid_list.sort()
            
        # get base filename
        bfn = self._opts.bfn
        if self._opts.nbfn != None:
            bfn = self._opts.nbfn
        
        # write jinf file
        self._jrnl_info.resize(self._opts.njf, self._opts.jfs)
        self._jrnl_info.write(self._jdir, bfn)
        
        # write records
        if self._opts.vflag:
            print "* Transferring records to new journal files"
        fro = self._jrnl_info.get_jrnl_sblk_size_bytes()
        while len(rid_list) > 0:
            hdr = master_record_list[rid_list.pop(0)]
            rec = hdr.encode()
            pos = 0
            while pos < len(rec):
                if self._file == None or self._file.tell() >= self._jrnl_info.get_jrnl_file_size_bytes():
                    if self._file == None:
                        rid = hdr.rid
                    elif len(rid_list) == 0:
                        rid = 0
                    else:
                        rid = rid_list[0]
                    if not self._rotate_file(rid, fro):
                        raise jerr.JournalSpaceExceededError()
                if len(rec) - pos <= self._jrnl_info.get_jrnl_file_size_bytes() - self._file.tell():
                    self._file.write(rec[pos:])
                    self._fill_file(jrnl.Utils.size_in_bytes_to_blk(self._file.tell(),
                                                                    self._jrnl_info.get_jrnl_dblk_size_bytes()))
                    pos = len(rec)
                    fro = self._jrnl_info.get_jrnl_sblk_size_bytes()
                else:
                    flen = self._jrnl_info.get_jrnl_file_size_bytes() - self._file.tell()
                    self._file.write(rec[pos:pos + flen])
                    pos += flen
                    rem = len(rec) - pos
                    if rem <= self._jrnl_info.get_jrnl_data_size_bytes():
                        fro = (jrnl.Utils.size_in_bytes_to_blk(self._jrnl_info.get_jrnl_sblk_size_bytes() + rem,
                                                               self._jrnl_info.get_jrnl_dblk_size_bytes()))
                    else:
                        fro = 0
            self._rec_wr_cnt += 1
            self._file_rec_wr_cnt += 1
        self._fill_file(add_filler_recs = True)
        while self._rotate_file():
            pass
        
    def _fill_file(self, to_posn = None, add_filler_recs = False):
        """Fill a file to a known offset"""
        if self._file == None:
            return
        if add_filler_recs:
            nfr = int(jrnl.Utils.rem_bytes_in_blk(self._file, self._jrnl_info.get_jrnl_sblk_size_bytes()) /
                      self._jrnl_info.get_jrnl_dblk_size_bytes())
            if nfr > 0:
                self._filler_wr_cnt = nfr
                for i in range(0, nfr):
                    self._file.write("RHMx")
                    self._fill_file(jrnl.Utils.size_in_bytes_to_blk(self._file.tell(),
                                                                    self._jrnl_info.get_jrnl_dblk_size_bytes()))
            self._last_rec_fid = self._fnum
            self._last_rec_offs = self._file.tell()
        if to_posn == None:
            to_posn = self._jrnl_info.get_jrnl_file_size_bytes()
        elif to_posn > self._jrnl_info.get_jrnl_file_size_bytes():
            raise jerr.FillExceedsFileSizeError(to_posn, self._jrnl_info.get_jrnl_file_size_bytes())
        diff = to_posn - self._file.tell()
        self._file.write(str("\0" * diff))
        #DEBUG
        if self._file.tell() != to_posn:
            raise jerr.FillSizeError(self._file.tell(), to_posn)
        
    def _rotate_file(self, rid = None, fro = None):
        """Switch to the next logical file"""
        if self._file != None:
            self._file.close()
            if self._opts.vflag:
                if self._file_rec_wr_cnt == 0:
                    print "  (empty)"
                elif self._filler_wr_cnt == None:
                    print "  (%d records)" % self._file_rec_wr_cnt
                else:
                    print "  (%d records + %d filler(s))" % (self._file_rec_wr_cnt, self._filler_wr_cnt)
        if self._fnum == None:
            self._fnum = 0
            self._rec_wr_cnt = 0
        elif self._fnum == self._jrnl_info.get_num_jrnl_files() - 1:
            return False
        else:
            self._fnum += 1
        self._file_rec_wr_cnt = 0
        self._fname = os.path.join(self._jrnl_info.get_jrnl_dir(), "%s.%04x.jdat" %
                                   (self._jrnl_info.get_jrnl_base_name(), self._fnum))
        if self._opts.vflag:
            print "* Opening file %s" % self._fname,
        self._file = open(self._fname, "w")
        if rid == None or fro == None:
            self._fill_file()
        else:
            now = time.time()
            fhdr = jrnl.FileHdr(0, "RHMf", jrnl.Hdr.HDR_VER, int(jrnl.Hdr.BIG_ENDIAN), 0, rid)
            fhdr.init(self._file, 0, self._fnum, self._fnum, fro, int(now), 1000000000*(now - int(now)))
            self._file.write(fhdr.encode())
            self._fill_file(self._jrnl_info.get_jrnl_sblk_size_bytes())
        return True
    
    def _handle_old_files(self):
        """Push old journal down into a backup directory"""
        target_dir = self._jdir
        if not self._opts.npd:
            target_dir = os.path.join(self._jdir, self.BAK_DIR)
            if os.path.exists(target_dir):
                if self._opts.vflag:
                    print "* Pushdown directory %s exists, deleting content" % target_dir
                for fname in glob.glob(os.path.join(target_dir, "*")):
                    os.unlink(fname)
            else:
                if self._opts.vflag:
                    print "* Creating new pushdown directory %s" % target_dir
                os.mkdir(target_dir)
        
        if not self._opts.npd or self._opts.obfn != None:
            if self._opts.obfn != None and self._opts.vflag:
                print "* Renaming old journal files using base name %s" % self._opts.obfn
            # .jdat files
            for fname in glob.glob(os.path.join(self._jdir, "%s.*.jdat" % self._opts.bfn)):
                tbfn = os.path.basename(fname)
                if self._opts.obfn != None:
                    per1 = tbfn.rfind(".")
                    if per1 >= 0:
                        per2 = tbfn.rfind(".", 0, per1)
                        if per2 >= 0:
                            tbfn = "%s%s" % (self._opts.obfn, tbfn[per2:])
                os.rename(fname, os.path.join(target_dir, tbfn))
            # .jinf file
            self._jrnl_info.write(target_dir, self._opts.obfn)
            os.unlink(os.path.join(self._jdir, "%s.jinf" % self._opts.bfn))

    def _print_options(self):
        """Print program options"""
        if self._opts.vflag:
            print "Journal dir: %s" % self._jdir
            print "Options: Base filename: %s" % self._opts.bfn
            print "         New base filename: %s" % self._opts.nbfn
            print "         Old base filename: %s" % self._opts.obfn
            print "         Pushdown: %s" % self._opts.npd
            print "         No. journal files: %d" % self._opts.njf
            print "         Journal file size: %d 64kiB blocks" % self._opts.jfs
            print "         Show records flag: %s" % self._opts.rflag
            print "         Verbose flag: %s" % True
            print
    
    def _process_args(self):
        """Process the command-line arguments"""
        opt = optparse.OptionParser(usage="%prog [options] DIR", version="%prog 1.0")
        opt.add_option("-b", "--base-filename",
                      action="store", dest="bfn", default="JournalData",
                      help="Base filename for old journal files")
        opt.add_option("-B", "--new-base-filename",
                      action="store", dest="nbfn",
                      help="Base filename for new journal files")
        opt.add_option("-n", "--no-pushdown",
                      action="store_true", dest="npd",
                      help="Suppress pushdown of old files into \"bak\" dir; old files will remain in existing dir")
        opt.add_option("-N", "--num-jfiles",
                      action="store", type="int", dest="njf", default=8,
                      help="Number of files for new journal (%d-%d)" % (self.NUM_JFILES_MIN, self.NUM_JFILES_MAX))
        opt.add_option("-o", "--old-base-filename",
                      action="store", dest="obfn",
                      help="Base filename for old journal files")
        opt.add_option("-q", "--quiet",
                      action="store_true", dest="qflag",
                      help="Quiet (suppress all non-error output)")
        opt.add_option("-r", "--records",
                      action="store_true", dest="rflag",
                      help="Print remaining records and transactions")
        opt.add_option("-s", "--jfile-size-pgs",
                      action="store", type="int", dest="jfs", default=24,
                      help="Size of each new journal file in 64kiB blocks (%d-%d)" %
                           (self.JFILE_SIZE_PGS_MIN, self.JFILE_SIZE_PGS_MAX))
        opt.add_option("-v", "--verbose",
                      action="store_true", dest="vflag",
                      help="Verbose output")
        (self._opts, args) = opt.parse_args()
        if len(args) == 0:
            opt.error("No journal directory argument")
        elif len(args) > 1:
            opt.error("Too many positional arguments: %s" % args)
        if self._opts.qflag and self._opts.rflag:
            opt.error("Quiet (-q/--quiet) and record (-r/--records) options are mutually exclusive")
        if self._opts.qflag and self._opts.vflag:
            opt.error("Quiet (-q/--quiet) and verbose (-v/--verbose) options are mutually exclusive")
        if self._opts.njf != None and (self._opts.njf < self.NUM_JFILES_MIN or self._opts.njf > self.NUM_JFILES_MAX):
            opt.error("Number of files (%d) is out of range (%d-%d)" %
                     (self._opts.njf, self.NUM_JFILES_MIN, self.NUM_JFILES_MAX))
        if self._opts.jfs != None and (self._opts.jfs < self.JFILE_SIZE_PGS_MIN or
                                       self._opts.jfs > self.JFILE_SIZE_PGS_MAX):
            opt.error("File size (%d) is out of range (%d-%d)" %
                     (self._opts.jfs, self.JFILE_SIZE_PGS_MIN, self.JFILE_SIZE_PGS_MAX))
        if self._opts.npd != None and (self._opts.nbfn == None and self._opts.obfn == None):
            opt.error("If (-n/--no-pushdown) is used, then at least one of (-B/--new-base-filename) and"
                     " (-o/--old-base-filename) must be used.")
        self._jdir = args[0]
        if not os.path.exists(self._jdir):
            opt.error("Journal path \"%s\" does not exist" % self._jdir)
        self._print_options()

#==============================================================================
# main program
#==============================================================================

if __name__ == "__main__":
    R = Resize()
    try:
        R.run()
    except Exception, e:
        sys.exit(e)