management/python/bin/qpid-store-chk

#!/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 optparse, os, sys


#== class StoreChk ============================================================

class StoreChk(object):
    """
    This class:
     1. Reads a journal jinf file, and from its info:
     2. Analyzes the journal data files to determine which is the last to be written, then
     3. Reads and analyzes all the records in the journal files.
    The only public method is run() which kicks off the analysis.
    """
    
    def __init__(self):
        """Constructor"""
        # params
        self.opts = None
        
        self._jdir = None
        
        # recovery analysis objects
#        self._jrnl_info = None
#        self.jrnl_rdr = 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_anal = janal.JrnlAnalyzer(self._jrnl_info)
        self.jrnl_rdr = janal.JrnlReader(self._jrnl_info, self.jrnl_anal, self.opts.qflag, self.opts.rflag,
                                         self.opts.vflag)

    def run(self):
        """Run the store check"""
        if not self.opts.qflag:
            print self._jrnl_info
            print self.jrnl_anal
        self.jrnl_rdr.run()
        self._report()
        
    def _report(self):
        """Print the results of the store check"""
        if not self.opts.qflag:
            print
            print " === REPORT ===="
            print
            print "Records:      %8d non-transactional" % \
                  (self.jrnl_rdr.get_msg_cnt() - self.jrnl_rdr.get_txn_msg_cnt())
            print "              %8d transactional" % self.jrnl_rdr.get_txn_msg_cnt()
            print "              %8d total" % self.jrnl_rdr.get_msg_cnt()
            print
            print "Transactions: %8d aborts" % self.jrnl_rdr.get_abort_cnt()
            print "              %8d commits" % self.jrnl_rdr.get_commit_cnt()
            print "              %8d total" % (self.jrnl_rdr.get_abort_cnt() + self.jrnl_rdr.get_commit_cnt())
            print
            if self.jrnl_rdr.emap().size() > 0:
                print "Remaining enqueued records (sorted by rid): "
                rid_list = self.jrnl_rdr.emap().rids()
                rid_list.sort()
                for rid in rid_list:
                    l = self.jrnl_rdr.emap().get(rid)
                    locked = ""
                    if l[2]:
                        locked += " (locked)"
                    print "  fid=%d %s%s" % (l[0], l[1], locked)
                print "WARNING: Enqueue-Dequeue mismatch, %d enqueued records remain." % self.jrnl_rdr.emap().size()
            else:
                print "No remaining enqueued records found (emap empty)."
            print
            if self.jrnl_rdr.tmap().size() > 0:
                txn_rec_cnt = 0
                print "Incomplete transactions: "
                for xid in self.jrnl_rdr.tmap().xids():
                    jrnl.Utils.format_xid(xid)
                    recs = self.jrnl_rdr.tmap().get(xid)
                    for l in recs:
                        print "  fid=%d %s" % (l[0], l[1])
                    print " Total: %d records for %s" % (len(recs), jrnl.Utils.format_xid(xid))
                    print
                    txn_rec_cnt += len(recs)
                print "WARNING: Incomplete transactions found, %d xids remain containing a total of %d records." % \
                      (self.jrnl_rdr.tmap().size(), txn_rec_cnt)
            else:
                print "No incomplete transactions found (tmap empty)."
            print
            print "%d enqueues, %d journal records processed." % \
                  (self.jrnl_rdr.get_msg_cnt(), self.jrnl_rdr.get_rec_cnt())
    
        
    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("-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 all records and transactions (including consumed/closed)")
        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")
        self._jdir = args[0]
        if not os.path.exists(self._jdir):
            opt.error("Journal path \"%s\" does not exist" % self._jdir)


#== class CsvStoreChk =========================================================

class CsvStoreChk(StoreChk):
    """
    This class, in addition to analyzing a journal, can compare the journal footprint (ie enqueued/dequeued/transaction
    record counts) to expected values from a CSV file. This can be used for additional automated testing, and is
    currently in use in the long store tests for journal encode testing.
    """
    
    # CSV file cols
    TEST_NUM_COL = 0
    NUM_MSGS_COL = 5
    MIN_MSG_SIZE_COL = 7
    MAX_MSG_SIZE_COL = 8
    MIN_XID_SIZE_COL = 9
    MAX_XID_SIZE_COL = 10
    AUTO_DEQ_COL = 11
    TRANSIENT_COL = 12
    EXTERN_COL = 13
    COMMENT_COL = 20
    
    def __init__(self):
        """Constructor"""
        StoreChk.__init__(self)

        # csv params
        self.num_msgs = None
        self.msg_len = None
        self.auto_deq = None
        self.xid_len = None
        self.transient = None
        self.extern = None
        
        self._warning = []
        
        self.jrnl_rdr.set_callbacks(self, CsvStoreChk._csv_pre_run_chk, CsvStoreChk._csv_enq_chk,
                                    CsvStoreChk._csv_deq_chk, CsvStoreChk._csv_txn_chk, CsvStoreChk._csv_post_run_chk)
        self._get_csv_test()

    def _get_csv_test(self):
        """Get a test from the CSV reader"""
        if self.opts.csvfn != None and self.opts.tnum != None:
            tparams = self._read_csv_file(self.opts.csvfn, self.opts.tnum)
            if tparams == None:
                print "ERROR: Test %d not found in CSV file \"%s\"" % (self.opts.tnum, self.opts.csvfn)
                sys.exit(1)
            self.num_msgs = tparams["num_msgs"]
            if tparams["min_size"] == tparams["max_size"]:
                self.msg_len = tparams["max_size"]
            else:
                self.msg_len = 0
            self.auto_deq = tparams["auto_deq"]
            if tparams["xid_min_size"] == tparams["xid_max_size"]:
                self.xid_len = tparams["xid_max_size"]
            else:
                self.xid_len = 0
            self.transient = tparams["transient"]
            self.extern = tparams["extern"]

    def _read_csv_file(self, filename, tnum):
        """Read the CSV test parameter file"""
        try:
            csvf = open(filename, "r")
        except IOError:
            print "ERROR: Unable to open CSV file \"%s\"" % filename
            sys.exit(1)
        for line in csvf:
            str_list = line.strip().split(",")
            if len(str_list[0]) > 0 and str_list[0][0] != "\"":
                try:
                    if (int(str_list[self.TEST_NUM_COL]) == tnum):
                        return { "num_msgs": int(str_list[self.NUM_MSGS_COL]),
                                 "min_size": int(str_list[self.MIN_MSG_SIZE_COL]),
                                 "max_size": int(str_list[self.MAX_MSG_SIZE_COL]),
                                 "auto_deq": not (str_list[self.AUTO_DEQ_COL] == "FALSE" or
                                                  str_list[self.AUTO_DEQ_COL] == "0"),
                                 "xid_min_size": int(str_list[self.MIN_XID_SIZE_COL]),
                                 "xid_max_size": int(str_list[self.MAX_XID_SIZE_COL]),
                                 "transient": not (str_list[self.TRANSIENT_COL] == "FALSE" or
                                                   str_list[self.TRANSIENT_COL] == "0"),
                                 "extern": not (str_list[self.EXTERN_COL] == "FALSE" or
                                                str_list[self.EXTERN_COL] == "0"),
                                 "comment": str_list[self.COMMENT_COL] }
                except Exception:
                    pass
        return None
        
    def _process_args(self):
        """Process 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("-c", "--csv-filename",
                      action="store", dest="csvfn",
                      help="CSV filename containing test parameters")
        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 all records and transactions (including consumed/closed)")
        opt.add_option("-t", "--test-num",
                      action="store", type="int", dest="tnum",
                      help="Test number from CSV file - only valid if CSV file named")
        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")
        self._jdir = args[0]
        if not os.path.exists(self._jdir):
            opt.error("Journal path \"%s\" does not exist" % self._jdir)
        
    # Callbacks for checking against CSV test parameters. Return False if ok, True to raise error.
    
    #@staticmethod
    def _csv_pre_run_chk(csv_store_chk):
        """Check performed before a test runs"""
        if csv_store_chk.num_msgs == None:
            return
        if csv_store_chk.jrnl_anal.is_empty() and csv_store_chk.num_msgs > 0:
            raise jerr.AllJrnlFilesEmptyCsvError(csv_store_chk.get_opts().tnum, csv_store_chk.num_msgs)
        return False
    _csv_pre_run_chk = staticmethod(_csv_pre_run_chk)
    
    #@staticmethod
    def _csv_enq_chk(csv_store_chk, hdr):
        """Check performed before each enqueue operation"""
        #if csv_store_chk.num_msgs == None: return
        # 
        if csv_store_chk.extern != None:
            if csv_store_chk.extern != hdr.extern:
                raise jerr.ExternFlagCsvError(csv_store_chk.opts.tnum, csv_store_chk.extern)
            if hdr.extern and hdr.data != None:
                raise jerr.ExternFlagWithDataCsvError(csv_store_chk.opts.tnum)
        if csv_store_chk.msg_len != None and csv_store_chk.msg_len > 0 and hdr.data != None and \
           len(hdr.data) != csv_store_chk.msg_len:
            raise jerr.MessageLengthCsvError(csv_store_chk.opts.tnum, csv_store_chk.msg_len, len(hdr.data))
        if csv_store_chk.xid_len != None and csv_store_chk.xid_len > 0 and len(hdr.xid) != csv_store_chk.xid_len:
            raise jerr.XidLengthCsvError(csv_store_chk.opts.tnum, csv_store_chk.xid_len, len(hdr.xid))
        if csv_store_chk.transient != None and hdr.transient != csv_store_chk.transient:
            raise jerr.TransactionCsvError(csv_store_chk.opts.tnum, csv_store_chk.transient)
        return False
    _csv_enq_chk = staticmethod(_csv_enq_chk)
    
    #@staticmethod
    def _csv_deq_chk(csv_store_chk, hdr):
        """Check performed before each dequeue operation"""
        if csv_store_chk.auto_deq != None and not csv_store_chk.auto_deq:
            raise jerr.JWarning("[CSV %d] WARNING: Dequeue record rid=%d found in non-dequeue test - ignoring." %
                                (csv_store_chk.opts.tnum, hdr.rid))
            #self._warning.append("[CSV %d] WARNING: Dequeue record rid=%d found in non-dequeue test - ignoring." %
            #                     (csv_store_chk.opts.tnum, hdr.rid))
        return False
    _csv_deq_chk = staticmethod(_csv_deq_chk)
    
    #@staticmethod
    def _csv_txn_chk(csv_store_chk, hdr):
        """Check performed before each transaction commit/abort"""
        return False
    _csv_txn_chk = staticmethod(_csv_txn_chk)

    #@staticmethod
    def _csv_post_run_chk(csv_store_chk):
        """Cehck performed after the completion of the test"""
        # Exclude this check if lastFileFlag is set - the count may be less than the number of msgs sent because
        # of journal overwriting
        if csv_store_chk.num_msgs != None and not csv_store_chk.jrnl_rdr.is_last_file() and \
           csv_store_chk.num_msgs != csv_store_chk.jrnl_rdr.get_msg_cnt():
            raise jerr.NumMsgsCsvError(csv_store_chk.opts.tnum, csv_store_chk.num_msgs,
                                       csv_store_chk.jrnl_rdr.get_msg_cnt())
        return False
    _csv_post_run_chk = staticmethod(_csv_post_run_chk)

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

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