dbsrv.py

Go to the documentation of this file.

#!/usr/bin/env python
"""
dbsrv : MySQL Server Utilities
================================

A more admin centric version of sibling *db.py* with advanced features, including:

* on server optimizations such as `select ... into outfile`
  taking advantage of the situation when the mysql client and
  server are on the same node.

* partitioned dump/load for dealing with very large tables and incremental backups

* implicit DB addressing without a *~/.my.cnf* section allowing handling of multiple databases all 
  from the same server via comma delimited names or regular expressions

* despite coming from NuWa it does not need the NuWa environment, system python with MySQLdb is OK

TODO
-----

#. checking the digests on the target and sending notification emails

#. test dumplocal when partitionsize is an exact factor of table size
#. warnings or asserts when using partitioned dumplocal with disparate table sizes 


Usage
------

::

   ./dbsrv.py tmp_ligs_offline_db_0 databases
   ./dbsrv.py tmp_ligs_offline_db_0 tables
   ./dbsrv.py tmp_ligs_offline_db_0 dumplocal  --where "SEQNO < 100" 
 
Similar to *db.py* the first argument can be a *~/.my.cnf* section name.
Differently to *db.py* it can also simply be a database name which
does not have a corresponding config section.  

In this implicit case the other connection pararameters are obtained 
from the so called *home* section.  Normally the *home* section is "loopback" 
indicating an on server connection.
The *home* section must point to the `information_schema` database. 

When the `--home` option is used databases on remote servers can
be accessed without having config sections for them all.

Comparing DB via partitioned dump
-----------------------------------

Three table dumps skipping the crashed table in order to compare:

* `dybdb1_ligs.tmp_ligs_offline_db_dybdb1`  original on dybdb1
* `dybdb2_ligs.channelquality_db_dybdb2`    recovered on dybdb2
* `loopback.channelquality_db_belle7`   recovered onto belle7 from hotcopy created on belle1

Invoked from cron for definiteness, and ability to leave running for a long time::

    07 17 * * * ( $DYBPYTHON_DIR/dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home dybdb1_ligs tmp_ligs_offline_db_dybdb1 dumplocal /tmp/cq/tmp_ligs_offline_db_dybdb1  --partition --partitioncfg 10000,0,33 ) >  $CRONLOG_DIR/dbsrv_dump_tmp_ligs_offline_db_dybdb1.log 2>&1 
    52 18 * * * ( $DYBPYTHON_DIR/dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home dybdb2_ligs channelquality_db_dybdb2 dumplocal /tmp/cq/channelquality_db_dybdb2 --partition --partitioncfg 10000,0,33 ) > $CRONLOG_DIR/dbsrv_dump_channelquality_db_dybdb2.log 2>&1 
    28 20 * * * ( $DYBPYTHON_DIR/dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home loopback channelquality_db_belle7 dumplocal /tmp/cq/channelquality_db_belle7 --partition --partitioncfg 10000,0,33 ) > $CRONLOG_DIR/dbsrv_dump_channelquality_db_belle7.log 2>&1 

.. warning:: `--partitioncfg` has now been split into `--partitionsize` and `--partitionrange`


Dump speed:

#. remote dumps from dybdb1/dybdb2 to belle7 take approx 165s for each chunk. Thus ~90min for all. 
#. local dumps on belle7 take approx 20s for each chunk. Thus ~11min for all.

diffing the dumped partitions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

For the first two all but the partial chunk match.  

Range of partition dirs to diff controlled by envvar::

    [blyth@belle7 DybPython]$ RANGE=0,10 ./diff.py  /tmp/cq/tmp_ligs_offline_db_dybdb1/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    [blyth@belle7 DybPython]$ RANGE=10,20 ./diff.py  /tmp/cq/tmp_ligs_offline_db_dybdb1/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    [blyth@belle7 DybPython]$ RANGE=20,30 ./diff.py  /tmp/cq/tmp_ligs_offline_db_dybdb1/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    [blyth@belle7 DybPython]$ RANGE=30,33 ./diff.py  /tmp/cq/tmp_ligs_offline_db_dybdb1/10000 /tmp/cq/channelquality_db_dybdb2/10000      ## see diff.py for the output from these

    [blyth@belle7 DybPython]$ RANGE=0,10 ./diff.py  /tmp/cq/channelquality_db_belle7/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    [blyth@belle7 DybPython]$ RANGE=10,20 ./diff.py  /tmp/cq/channelquality_db_belle7/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    [blyth@belle7 DybPython]$ RANGE=20,30 ./diff.py  /tmp/cq/channelquality_db_belle7/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    [blyth@belle7 DybPython]$ RANGE=30,33 ./diff.py  /tmp/cq/channelquality_db_belle7/10000 /tmp/cq/channelquality_db_dybdb2/10000   

oops a difference, but its just different formatting of 0.0001 or 1e-04
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

::

    [blyth@belle7 DybPython]$  RANGE=10,20 ./diff.py  /tmp/cq/channelquality_db_belle7/10000 /tmp/cq/channelquality_db_dybdb2/10000   
    2013-06-07 17:58:06,933 __main__ INFO     rng ['10', '11', '12', '13', '14', '15', '16', '17', '18', '19'] 
    2013-06-07 17:58:26,526 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/10 /tmp/cq/channelquality_db_dybdb2/10000/10 => 0 
    2013-06-07 17:58:44,896 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/11 /tmp/cq/channelquality_db_dybdb2/10000/11 => 0 
    2013-06-07 17:59:04,360 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/12 /tmp/cq/channelquality_db_dybdb2/10000/12 => 0 
    2013-06-07 17:59:22,531 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/13 /tmp/cq/channelquality_db_dybdb2/10000/13 => 0 
    2013-06-07 17:59:42,205 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/14 /tmp/cq/channelquality_db_dybdb2/10000/14 => 0 
    2013-06-07 18:00:00,385 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/15 /tmp/cq/channelquality_db_dybdb2/10000/15 => 0 
    2013-06-07 18:00:20,000 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/16 /tmp/cq/channelquality_db_dybdb2/10000/16 => 0 
    2013-06-07 18:00:38,198 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/17 /tmp/cq/channelquality_db_dybdb2/10000/17 => 0 
    2013-06-07 18:00:38,704 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/18 /tmp/cq/channelquality_db_dybdb2/10000/18 => 1
    Files /tmp/cq/channelquality_db_belle7/10000/18/DqChannel.csv and /tmp/cq/channelquality_db_dybdb2/10000/18/DqChannel.csv differ
     
    2013-06-07 18:00:56,602 __main__ INFO     diff -r --brief /tmp/cq/channelquality_db_belle7/10000/19 /tmp/cq/channelquality_db_dybdb2/10000/19 => 0 
    [blyth@belle7 DybPython]$ 
    [blyth@belle7 DybPython]$ 
    [blyth@belle7 DybPython]$ 
    [blyth@belle7 DybPython]$ diff /tmp/cq/channelquality_db_belle7/10000/18/DqChannel.csv  /tmp/cq/channelquality_db_dybdb2/10000/18/DqChannel.csv 
    1196930c1196930
    < 186235,2,28473,7,67175938,0.0001,7.35714,3.39868,-1,-1
    ---
    > 186235,2,28473,7,67175938,1e-04,7.35714,3.39868,-1,-1
    ...

Commands
-----------

summary
~~~~~~~~~

Providea a summary of table counts and update times in all selected databases.
The DB names are specified by comma delimited OR Regexp string arguments specifying the DB names. 
::

    ./dbsrv.py               tmp_ligs_offline_db_\\d summary           

               # local home, requires "loopback" config section pointing to information_schema DB

    ./dbsrv.py --home dybdb1 tmp_\\S* summary                          

               # remote home,  requires "dybdb1" config section pointing to information_schema DB

TODO:

Check handling of section names the same as DB names on different nodes, as the section config will trump the dbname ? 
BUT *home* config host matching should trip asserts ? 


dumplocal
~~~~~~~~~

The DB tables are dumped as *.csv* files and separate *.schema* files containing table creation SQL.
Without a directory argument the dumps are writes beneath the `--backupfold` controllable directory, such as `/var/dbbackup/dbsrv`

::

    [blyth@belle7 DybPython]$ ./dbsrv.py  tmp_ligs_offline_db_0 dumplocal --where 'SEQNO <= 100'    
    2013-06-13 16:49:38,152 __main__ INFO     partition_dumplocal___ SEQNO <= 100 writing /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_0/DqChannel.csv 
    2013-06-13 16:49:38,578 __main__ INFO     partition_dumplocal___ SEQNO <= 100 writing /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_0/DqChannel.csv took  0.39 seconds 
    ...

    [blyth@belle7 DybPython]$ ./dbsrv.py  tmp_ligs_offline_db_0 dumplocal /tmp/check/tmp_ligs_offline_db_0 --where 'SEQNO <= 100' 
    2013-06-13 16:50:49,003 __main__ WARNING  using basedir /tmp/check/tmp_ligs_offline_db_0 different from standard /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_0 
    2013-06-13 16:50:49,031 __main__ INFO     partition_dumplocal___ SEQNO <= 100 writing /tmp/check/tmp_ligs_offline_db_0/DqChannel.csv 
    2013-06-13 16:50:49,203 __main__ INFO     partition_dumplocal___ SEQNO <= 100 writing /tmp/check/tmp_ligs_offline_db_0/DqChannel.csv took  0.17 seconds 
    ...  

.. warning:: When there are databases of the same name on multiple nodes it is useful to include the names of the node in the section name


loadlocal
~~~~~~~~~~~

When doing a load into a database to be created use `--DB_DROP_CREATE` option::

    [blyth@belle7 DybPython]$ ./dbsrv.py tmp_ligs_offline_db_5 loadlocal ~/tmp_ligs_offline_db_0  -l debug --DB_DROP_CREATE 

Typically when loading a database name change in needed, in this case the directory and new section name must be given::

    [blyth@belle7 DybPython]$ ./dbsrv.py tmp_ligs_offline_db_50 loadlocal /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_0 --DB_DROP_CREATE  
    DROP and reCREATE database tmp_ligs_offline_db_50 loosing all tables contained ? Enter "YES" to proceed : YES
    2013-06-13 16:58:41,491 __main__ WARNING  using basedir /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_0 different from standard /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_50 
    2013-06-13 16:58:41,499 __main__ WARNING  creating table DqChannel from schema file /var/dbbackup/dbsrv/belle7.nuu.edu.tw/tmp_ligs_offline_db_0/DqChannel.schema 
    ...


partitioned loadlocal
~~~~~~~~~~~~~~~~~~~~~~~~

NB when restoring need to do a name change, so it is neccesary to specify the source directory as an argument  
::

     [root@cms01 DybPython]# dbsrv channelquality_db_restored loadlocal /data/var/dbbackup/dbsrv/dybdb2.ihep.ac.cn/channelquality_db_dybdb2 --partition --extract -l debug --DB_DROP_CREATE  -C
                             ##  initial run, creating the DB  from 32 partitions took ~100 min
  
     [root@cms01 DybPython]# dbsrv channelquality_db_restored loadlocal /data/var/dbbackup/dbsrv/dybdb2.ihep.ac.cn/channelquality_db_dybdb2 --partition --extract  -K
                             ##  quick re-run, notices nothing to do and completes in a few seconds

     [blyth@cms01 ~]$ type dbsrv    # function to nab the NuWa python MySQLdb, as yum is being uncooperative on cms01
     dbsrv is a function
     dbsrv () 
     { 
        local python=/data/env/local/dyb/trunk/external/Python/2.7/i686-slc4-gcc34-dbg/bin/python;
        export PYTHONPATH=/data/env/local/dyb/trunk/NuWa-trunk/../external/mysql_python/1.2.3_mysql5.0.67_python2.7/i686-slc4-gcc34-dbg/lib/python2.7/site-packages;
        LD_LIBRARY_PATH=/data/env/local/dyb/trunk/NuWa-trunk/../external/mysql/5.0.67/i686-slc4-gcc34-dbg/lib/mysql:$LD_LIBRARY_PATH;
        LD_LIBRARY_PATH=/data/env/local/dyb/trunk/NuWa-trunk/../external/mysql_python/1.2.3_mysql5.0.67_python2.7/i686-slc4-gcc34-dbg/lib:$LD_LIBRARY_PATH;
        export LD_LIBRARY_PATH;
        $python -c "import MySQLdb";
        $python ~blyth/DybPython/dbsrv.py $*
     }

Test run on cms01 chugging along at ~3 min per 10k partition, at 32 partitions estimate ~100 min to complete

::

      [blyth@belle7 DybPython]$ ./dbsrv.py  channelquality_db_restored loadlocal  /var/dbbackup/dbsrv/dybdb2.ihep.ac.cn/channelquality_db_dybdb2 --partition --extract -l debug --DB_DROP_CREATE  -C



Partitioned Commands
----------------------

The partitioning relies on these options:

`--partition`
          switches on partitioning, default False
`--partitionkey`
          default "SEQNO,0", corresponding to the key name and its position in CSV dumps
`--partitioncfg`
          **NOW RPLACED WITH THE BELOW TWO OPTIONS**
          default "10000,0,33", the three integers specify the number of keys in each chunk `10000` 
          and the range of chunks `range(0,33)` ie 0 to 32
`--partitionsize`
          default "10000", specify the number of keys in each chunk
`--partitionrange`
          default of None, meaning all partitions. 
          If specified as eg "0,33" it restricts to a range of partition indices `range(0,33)`
--partitionlast`
          **NOW DEPRECATED** This the last partition is now auto determined, to allow daily cron running 
          default None, when set to an integer string eg "32" this is used to identifiy the index 
          of the last incomplete partition

For dump and load to refer to the same partition set, requires the same chunk size 
(and partition key although this is not checked).

partitioned loadlocal
~~~~~~~~~~~~~~~~~~~~~~~

From cron::

    DYBPYTHON_DIR=/data1/env/local/dyb/NuWa-trunk/dybgaudi/DybPython/python/DybPython
    03 20 * * * ( $DYBPYTHON_DIR/dbsrv.py channelquality_db_0 loadlocal  /tmp/cq/channelquality_db --partition --DB_DROP_CREATE -C ) > $CRONLOG_DIR/dbsrv_load_.log 2>&1

quick partitioning test
~~~~~~~~~~~~~~~~~~~~~~~~

For fast dump/load testing use small chunks and range of partitions::

    ./dbsrv.py tmp_ligs_offline_db_0 dumplocal /tmp/pp/tmp_ligs_offline_db_0 --partition --partitionsize 10 --partitionrange 0,5
    ./dbsrv.py tmp_ligs_offline_db_5 loadlocal /tmp/pp/tmp_ligs_offline_db_0 --partition --partitionsize 10 --partitionrange 0,5 --DB_DROP_CREATE -C 


Archiving and transfers to remote node
-----------------------------------------

Controlled via options:

`-a/--archive`
          switch on archive creation 

`-x/--extract`
          switch on archive extraction 

`--backupfold`
          default /var/dbbackup/dbsrv, the location of backup dumps and tarballs

`-T/--transfer`
          switch on remote transfer of archives, must be used together with `-a/--archive` and the *dumplocal* command to be effective
   
`--transfercfg`
          configures the remote node and possible a directory prefix, that is prepended infront of the `backupfold`


For example the below command dumps partitions 0,1 and 2, creates archive tarballs and transfers them to the remote node configured::

     ./dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home loopback channelquality_db_belle7 dumplocal /tmp/cq/channelquality_db_belle7 --partition --partitionrange 0,3 -aT

The local and remote tarball paths are the same, with no transfercfg prefix specified, namely::

     /var/dbbackup/dbsrv/belle7.nuu.edu.tw/channelquality_db_belle7/10000_0.tar.gz 


Transfer Optimization 
~~~~~~~~~~~~~~~~~~~~~~~

A small .dna sidecar to the tarballs is used for tarball content identification.
When a rerun of the transfer is made, the sidecar DNA is first checked to see
if the remote node already holds the tarball.

This means that only newly reached partitions are archived and transferred.  
The `last` incomplete partition will typically be transferred every time as it will
have a different content causing the DNA mismatch to trigger a re-transfer.


Full archive/transfer cron test from belle7 to belle1 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

To prepare the remote node just need to create and set ownership of `backupfold` eg  `/var/dbbackup/dbsrv` 
and ensure keyed ssh access is working

::

    DYBPYTHON_DIR=/data1/env/local/dyb/NuWa-trunk/dybgaudi/DybPython/python/DybPython
    DBSRV_REMOTE_NODE=N1
    35 18 * * * ( $DYBPYTHON_DIR/dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home loopback channelquality_db_belle7 dumplocal --partition --archive --transfer ) > $CRONLOG_DIR/dbsrv_pat_channelquality_db_belle7.log 2>&1  


Installation on dybdb2
------------------------

Prepare target node 
~~~~~~~~~~~~~~~~~~~~~

The administrator of target node needs to prepare a folder for the archives::

    [blyth@cms01 ~]$ sudo mkdir /data/var/dbbackup/dbsrv
    [blyth@cms01 ~]$ sudo chown -R dayabayscp.dayabayscp /data/var/dbbackup/dbsrv

Setup mysql config at source
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

The config file :file:`~/.my.cnf` needs two sections "loopback" and "channelquality_db_dybdb2"::

    [loopback]
    host      = 127.0.0.1
    database  = information_schema
    user      = root
    password  = ***

    [channelquality_db_dybdb2]
    host      = 127.0.0.1
    database  = channelquality_db
    user      = root
    password  = ***

SSH environment configuration
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

The script runs *scp* commands internally that require:

* `ssh-agent` process to be running and authenticated 
* public keys of source node to be appended to `.ssh/authorized_keys2` of target 
* :envvar:`SSH_AUTH_SOCK` to be defined.  

When run from cron the envvar is typically not present.  
In order to define this the :file:`~/.ssh-agent-info-$NODE_TAG` 
is parsed by the `sshenv()` from common.py.

This file is created by the **env** function `ssh--agent-start` which is used 
following reboots to start and authenticate the ssh agent process. 

* http://belle7.nuu.edu.tw/e/base/ssh/

Get DybPython from dybsvn
~~~~~~~~~~~~~~~~~~~~~~~~~~~

::

    cd 
    svn co http://dayabay.ihep.ac.cn/svn/dybsvn/dybgaudi/trunk/DybPython/python/DybPython

Despite coming from dybsvn the *dbsrv.py* script does not need the NuWa environment. Just
the MySQLdb extension in the system python should be adequate.


Quick Interactive Test
~~~~~~~~~~~~~~~~~~~~~~~

Configuring 5 small 100 SEQNO partitions allows the machinery to be quickly tested::

    cd DybPython
    ./dbsrv.py channelquality_db_dybdb2 dumplocal --partition --partitionsize 100 --partitionrange 0,5 --archive --transfer  


CRON commandline
~~~~~~~~~~~~~~~~~~

::

    ## NB no DBSRV_REMOTE_NODE is needed, the default of S:/data is appropriate
    DYBPYTHON_DIR=/root/DybPython
    CRONLOG_DIR=/root/cronlog
    NODE_TAG=D2
    #
    42 13 * * * ( $DYBPYTHON_DIR/dbsrv.py channelquality_db_dybdb2 dumplocal --partition --archive --transfer ) > $CRONLOG_DIR/dbsrv_channelquality_db_dybdb2.log 2>&1 


A do-nothing run, when there are no new partitions to dump/archive/transfer takes about 4 mins and uses little resources.
When there are new completed partitions to archive and transfer, the default chunk size of 10000 SEQNO leads to tarballs 
of only 35M (maybe 70M when move for all 4 tables) resulting in rapid transfers.

Although new completed partitions might be reached perhaps every ~10 days with the 10k chunks, a daily 
transfer is still recommended in order to backup the last incomplete partition and also in order that 
issues with the transfer are rapidly identified and resolved.



Transfer Monitoring
~~~~~~~~~~~~~~~~~~~

Implemented using *valmon.py* with *digestpath.py*. Valmon needs to run as a daily 
cronjob on the remote node. Configure with **dbsrvmon** section::

    % ~/.env.cnf blyth@belle1.nuu.edu.tw 
    [dbsrvmon]
    tn = channelquality_db 
    chdir = /var/dbbackup/dbsrv/belle7.nuu.edu.tw/channelquality_db_belle7/archive/10000 
    return = dict 
    dbpath = ~/.env/dbsrvmon.sqlite 
    cmd = digestpath.py 
    note = stores the dict returned by the command as a string in the DB without interpretation 
    valmon_version = 0.2 
    constraints = ( tarball_count >= 34, dna_mismatch == 0, age < 86400 , age < 1000, ) 

Tested on belle1::

    [blyth@belle1 e]$ valmon.py -s dbsrvmon ls
    2013-06-17 11:48:01,515 env.db.valmon INFO     /home/blyth/env/bin/valmon.py -s dbsrvmon ls
    2013-06-17 11:48:01,520 env.db.valmon WARNING  no email section configures and no MAILTO envvar, NOTIFICATION WILL FAIL
    2013-06-17 11:48:01,521 env.db.valmon INFO     arg ls
    ('2013-06-13T19:46:01', 5.5278148651123047, "{'dna_match': 34, 'lookstamp': 1371123961.7826331, 'dna_mismatch': 0, 'tarball_count': 34, 'age': 9030.7826330661774, 'lastchange': 1371114931, 'dna_missing': 0}", 0.0, 0)
    ('2013-06-13T19:54:06', 5.8677470684051514, "{'dna_match': 34, 'lookstamp': 1371124446.7869501, 'dna_mismatch': 0, 'tarball_count': 34, 'age': 9515.7869501113892, 'lastchange': 1371114931, 'dna_missing': 0}", 0.0, 0)



Obtain the backup tarballs
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

As of Dec 24 2013 there are 54 tarballs of 43M each, corresponding to 2322M total. scp them using the
scponly account on cms01. Qiumei/Simon can provide the password::

    dir=/data/var/dbbackup/dbsrv/dybdb2.ihep.ac.cn/channelquality_db_dybdb2/archive/
    mkdir -p $dir && cd $dir         
    scp -r dayabayscp@cms01.phys.ntu.edu.tw:/data/var/dbbackup/dbsrv/dybdb2.ihep.ac.cn/channelquality_db_dybdb2/archive/10000 .


Partioned dump usage
-----------------------

Full backups are impractical for 10G tables.

Partitioned dumping is attactive for backups of such large tables, 
as just new partitions need to be dumped on each invokation. 

For scp transfers would need to create tarfiles for each partition 
with dna sidecars, and add a transfer subcommand with option controlled remote node. 
Clearly via dna checking would allow only new partitions to be transfereed.



System python API warning
--------------------------

Careful regarding PYTHONPATH, when mixing a NuWa PYTHONPATH with a system python get API RuntimeWarning::

    [blyth@belle7 DybPython]$ /usr/bin/python dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home dybdb2_ligs   channelquality_db_dybdb2 dumplocal /tmp/cq/channelquality_db_dybdb2  --partition --partitioncfg 10,0,1 
    /data1/env/local/dyb/external/mysql_python/1.2.3_mysql5.0.67_python2.7/i686-slc5-gcc41-dbg/lib/python2.7/site-packages/MySQLdb/__init__.py:19: RuntimeWarning: Python C API version mismatch for module _mysql: This Python has API version 1012, module _mysql has version 1013.
      import _mysql
    2013-06-06 18:03:08,963 __main__ INFO     schema dir /tmp/cq/channelquality_db_dybdb2/10/_ exists already
    2013-06-06 18:03:08,963 __main__ INFO     /* 10-partition 1  /1  */ SEQNO >= 1 and SEQNO <= 10 
    2013-06-06 18:03:09,165 __main__ INFO     checking prior csv dump /tmp/cq/channelquality_db_dybdb2/10/0  --partitioncfg 10,0,1 
    [blyth@belle7 DybPython]$ 

Avoiding the NuWa PYTHONPATH means are entirely system and avoid the RuntimeWarning::

    [blyth@belle7 DybPython]$ PYTHONPATH=  /usr/bin/python dbsrv.py -t DqChannel,DqChannelVld,DqChannelStatusVld --home dybdb2_ligs   channelquality_db_dybdb2 dumplocal /tmp/cq/channelquality_db_dybdb2  --partition --partitioncfg 10,0,1 
    2013-06-06 18:04:58,078 __main__ INFO     schema dir /tmp/cq/channelquality_db_dybdb2/10/_ exists already
    2013-06-06 18:04:58,078 __main__ INFO     /* 10-partition 1  /1  */ SEQNO >= 1 and SEQNO <= 10 
    2013-06-06 18:04:58,282 __main__ INFO     checking prior csv dump /tmp/cq/channelquality_db_dybdb2/10/0  --partitioncfg 10,0,1 
    [blyth@belle7 DybPython]$ 


Import Notes
-------------

#. keep to a minimum of imports for portability to server situation, ie do not rely on NuWa environment
#. MySQLdb optionality is to allows non MySQL-python nodes to autodoc

"""
import sys, os, logging, inspect, time, re, tarfile, platform, platform
from datetime import datetime
log = logging.getLogger(__name__)
from ConfigParser import ConfigParser, NoSectionError
from pprint import pformat
from tar import Tar
from common import sshenv



def _sorted(iterable, key=lambda _:_, reverse=False):
    """  
    sorted for py23, caution returns lists not tuples
    """ 
    temp = [(key(x), x) for x in iterable]
    temp.sort()
    if reverse:
        return [temp[i][1] for i in xrange(len(temp) - 1, -1, -1)] 
    return [t[1] for t in temp]

try:
    sorted
except NameError: 
    sorted = _sorted 



try:
    import MySQLdb
except ImportError:
    MySQLdb = None

try:
    import _mysql
except ImportError:
    _mysql = None

 
def attrs_(mod, names):
    """Access module constants is a version independant way """  
    if mod:
        return map( lambda _:getattr(mod, _), filter( lambda _:hasattr(mod,_), names.split() ) ) 
    else:
        return None 
string_types = attrs_( MySQLdb.constants.FIELD_TYPE , 'VARCHAR DATETIME STRING VAR_STRING' ) 


seconds = {}
def timing(func):
    def wrapper(*arg,**kw):
        '''source: http://www.daniweb.com/code/snippet368.html'''
        t1 = time.time()
        res = func(*arg,**kw)
        t2 = time.time()
        global seconds
        seconds[func.func_name] = (t2-t1)
        return res 
    return wrapper


class CSVFormat(list):
    """
    Provides the format string to create the CSV line
    appropriate for the field types of the low level query result description 
    It looks something like::

        %s,"%s","%s",%s,%s,%s,%s,%s,"%s","%s"

    Usage example::

        llc = _mysql.connection(...) 
        llc.query("select * from ...")
        result = llc.store_result()
        csvf = CSVFormat( result.describe() ) 
        for row in result.fetch_row(0):
            print str(csvf) % tuple(row)

    """

    def __str__(self):
        def field_fmt(fdesc):
            if fdesc[1] in string_types:
                return "\"%s\""
            return "%s"
        return ",".join( map(field_fmt, self) )    
 
class DB(object):
    def __init__(self, sect, opts=None, home=None):
        """
        :param sect: name of section in config file 
        :param opts: options
        :param home: DB instance


        Safety constraints on config to minimize accidents from config confusion.

        Initially required non-loopback section names and database names to be the same

        Loosening this to allow remote commands, by designating a "home" instance
        and requiring all other instances to match that one in everything but the 
        database name

        """
        pass 
        self.sect = sect
        self.opts = opts
        self.home = home
        self.backupdir = os.path.join(self.opts.backupfold, platform.node(), sect )   

        cnf = MyCnf("~/.my.cnf")
        dbc = cnf.mysqldb_pars(sect, home=home)   # NB sect is taken as the DB name if there is no such section in the config

        log.debug("sect %s home %s dbc %s " % (sect, home, repr(dbc)))
 
        if sect == "loopback":
            assert dbc['host'] in ('localhost','127.0.0.1')   # curious 127.0.0.1 not working on cms01 need localhost
        pass
        if home is None: 
            assert dbc['db'] in ("information_schema", "mysql")  # older mysql (eg on cms01 does not have information_schema DB ? OR its protected)
            assert opts.home == sect, "only the home instance is allowed an undefined home parameter " 
        else:
            log.debug("     dbc %s " % repr(dbc))
            log.debug("home.dbc %s " % repr(home.dbc))
            #qwns = "host user passwd"
            qwns = "host"
            for qwn in qwns.split():
                assert dbc.get(qwn,None) == home.dbc.get(qwn,None), "non \"home\" instances are constrained to match the \"home\" in everything but the db "

        self.is_local = dbc['host'] == '127.0.0.1'
        self.dbc = dbc
        self.cnf = cnf
        log.debug("connecting to %s " % dict(dbc, passwd="***"))
        try:  
            conn = MySQLdb.connect( **dbc )   # huh, version variation in accepted params
        except MySQLdb.Error, e: 
            raise Exception("Error %d: %s " % ( e.args[0], e.args[1] ) )

        llconn = _mysql.connect( **dbc )       

        self.conn = conn
        self.llconn = llconn
        self._size = None
        pass

        partitionmgr = PartitionMgr(opts, self)
        self.partitionmgr = partitionmgr


    def dispatch(self, *args, **kwa):
        if self.opts.partition:
            pfx = "partition_"
        else:
            pfx = ""

        cmd = pfx + args[0] + "___"    
        if hasattr(self, cmd ):
            getattr( self , cmd)( *args[1:], **kwa )   
        else:
            raise Exception("cmd %s not implemented " % cmd)

    def lscmd___(self, *args, **kwa ):
        for m in filter(lambda _:_[-3:] == '___', dir(self)):
            print m[:-3]  

    def database_drop_create(self, dbname):
        """
        :param dbname: name of the database to be dropped and recreated 
        """
        assert self.sect == "loopback" and self.sect != dbname , ( self.sect, dbname  )
        if self.opts.noconfirm:
            log.info("proceed with DB_DROP_CREATE of %(dbname)s without confirmation" % locals() )
        else:
            ans = raw_input("DROP and reCREATE database %(dbname)s loosing all tables contained ? Enter \"YES\" to proceed : " % locals()) 
            if ans != "YES":
                log.warn("skipping DROP CREATE ")
                return
       
        self("drop database if exists %(dbname)s " % locals() ) 
        self("create database %(dbname)s " % locals() ) 


    def docs( cls ):
        """
        collect the docstrings on command methods 
        identified by naming convention of ending with ___ 
        """
        mdoc = lambda m:getattr(m,'__doc__',None)
        mdocs  = [ dict(meth=k[:-3],doc=mdoc(v)) for k,v in [(k,v) for k,v in inspect.getmembers(cls) if k[-3:]=='___' and k[0] != '_' and mdoc(v)]]
        return "\n".join([ """ %(meth)s : %(doc)s """ % d for d in mdocs ])   
    docs = classmethod(docs)


    def execute_(self, cmd):
        cursor = self.conn.cursor(MySQLdb.cursors.DictCursor)
        cursor.execute( cmd )
        return cursor

    def fetchall(self, cmd ):
        cursor = self.execute_(cmd)
        rows = cursor.fetchall()
        self.count = cursor.rowcount
        cursor.close()
        return rows

    def _query_size(self):
        sql = "select round(sum((data_length+index_length-data_free)/1024/1024),2) as TOT_MB from information_schema.tables where table_schema = '%(db)s' " % self.dbc
        return float(self(sql)[0]['TOT_MB'])
    def _get_size(self):
        if self._size is None:
             self._size = self._query_size()
        return self._size 
    size = property(_get_size, doc="Size estimate of the DB in MB ") 

    def _get_databases(self):
        """
        This query gives fewer results than `show databases`, which demands skips to avoid errors in getting sizes 
        #skip = "hello hello2 other test_noname tmp_cascade_2 tmp_dbitest tmp_tmp_offline_db_2".split()  
        """
        sql = "select distinct(table_schema) from information_schema.tables"
        return map(lambda _:_['table_schema'],self(sql))
    databases = property(_get_databases, doc="List of database names obtained from information_schema.tables") 


    def _get_tables_infoschema(self):
        """
        """
        sql = "select distinct(table_name) from information_schema.tables where table_schema='%(db)s'" % self.dbc
        return map(lambda _:_['table_name'],self(sql))
    tables = property(_get_tables_infoschema, doc="List of table names obtained from information_schema.tables") 


    def _get_tables(self):
        """
        Older mysql does not have information_schema
        """
        sql = "show tables" 
        tables = []
        for d in self(sql):
            k,v = d.items()[0]
            tables.append(v)
        return tables
    tables = property(_get_tables, doc="List of table names obtained from show tables") 


    def _get_datadir(self):
        return self("select @@datadir as datadir")[0]['datadir']
    datadir = property(_get_datadir, doc="Query DB server to find the datadir, eg /var/lib/mysql/ OR /data/mysql/ ")
        
    def __call__(self, cmd):
        log.debug(cmd)
        t0 = time.time()
        ret = self.fetchall(cmd)
        t1 = time.time()
        self.lastseconds = t1 - t0 
        return ret


    def lsdatabases___(self, *args, **kwa ):
        """
        list databases
        """ 
        print "\n".join(self.databases)

    def lstables___(self, *args, **kwa ):
        """
        list tables
        """
        print "\n".join(self.tables)

    def summary___(self, *args, **kwa):
        """
        Present summary of tables in rst table format:

        ==============================  ==========  ==============================  ==============================
        TABLE_NAME                      TABLE_ROWS  CREATE_TIME                     CHECK_TIME                    
        ==============================  ==========  ==============================  ==============================
        DqChannel                       62126016    2013-05-30 18:52:51             2013-05-30 18:52:51           
        DqChannelStatus                 62126016    2013-05-30 18:17:42             2013-05-30 18:17:42           
        DqChannelStatusVld              323573      2013-05-30 18:52:44             None                          
        DqChannelVld                    323573      2013-05-30 19:34:55             None                          
        LOCALSEQNO                      3           2013-05-30 19:35:02             None                          
        ==============================  ==========  ==============================  ==============================
             
        """ 
        result = self("select TABLE_NAME, TABLE_ROWS, CREATE_TIME, UPDATE_TIME, CHECK_TIME  from information_schema.tables where table_schema = '%(db)s' " % self.dbc)
        kws = (('TABLE_NAME', 30), ('TABLE_ROWS', 10), ('CREATE_TIME', 30 ), ('CHECK_TIME',30),)
        print "\n".join(["",self.sect,self._rst_table( result, kws )])
   
    
    def _rst_table(self, result, kws, char="="):
        """
        :param result: interable providing result dicts 
        :param kws: sequence of 2-tuples providing result dict keys and presentation widths 
        :param char: 
        :return: multi line string rst table presentation 
        """ 
        fcol_ = lambda kw:"%s(%s)-%s%s" % ("%",kw[0],kw[1],"s")
        fmt = "  ".join(map(fcol_,kws) ) 

        # spell these out for ancient python 
        #mkr = fmt % dict((kw[0],char * kw[1]) for kw in kws)
        #lbl = fmt % dict((kw[0],kw[0]) for kw in kws)
        dmkr = {}
        dlbl = {}
        for kw in kws:
            dmkr[kw[0]] = char * kw[1] 
            dlbl[kw[0]] = kw[0] 
        mkr = fmt % dmkr
        lbl = fmt % dlbl
        bdy = [fmt % d for d in result]
        return "\n".join([mkr, lbl, mkr] + bdy + [mkr])


    def _get_utables(self):
        """
        """ 
        alltables = self._get_tables()
        if self.opts.tables is None:
             return alltables
        else:
             return filter(lambda t:t in alltables, self.opts.tables.split(","))
    utables = property(_get_utables, doc="List of tables to use in operations, when --tables option is used this can be a subset of all tables.")

    def show_create( self, table):
        return self("show create table %(table)s" % locals())[0]['Create Table']

    def _dumplocal_mkdir(self, outdir ):
        """
        :param outdir:
        """
        if not os.path.exists(outdir):
            os.makedirs(outdir)     # umask gets in the way of the mode argument here 
            os.chmod(outdir, 0777)  # needs to be writable by other so mysql user can write into it 
        assert os.path.isdir(outdir), outdir
        pass
    def _dumplocal_schema(self, outdir, table ):
        """
        :param outdir:
        :param table:
        """
        schema=os.path.join(outdir, "%s.schema" % table )
        sf = open(schema, "w")
        sf.write( self.show_create(table) )
        sf.close()
        pass
    def _dumplocal_table(self, outdir, table, where ):
        """
        :param outdir:
        :param table:
        :param where: 
        """
        outfile=os.path.join(outdir, "%s.csv" % table )
        log.info("_dumplocal_table %s writing %s " % (where,outfile) ) 
        pass

        if table == 'LOCALSEQNO':
            where = "TRUE"

        if self.is_local:
            log.debug("IntoOutFile")
            if os.path.exists(outfile):
                log.info("remove preexisting outfile %s " % outfile )
                os.remove(outfile) 
            io = IntoOutfile(table=table,where=where, outfile=outfile)
            self(str(io))
        else:
            log.debug("RemoteIntoOutFile")
            t0 = time.time()
            tf = open(outfile,"w")
            rio = RemoteIntoOutfile(table=table, where=where)
            self._write_csvdirect(rio, tf )   ## result of select is returned to python and thence formatted directly into csv, works remotely 
            tf.close()
            t1 = time.time()
            self.lastseconds = t1 - t0
        pass
        log.info("partition_dumplocal___ %s writing %s took %s seconds " % (where,outfile, "%5.2f" % self.lastseconds ) ) 
        pass

    def _write_csvdirect(self, select , tf ):
        """
        Adopt low level approach to avoid unnecessary conversions into 
        python types then back to string and the associated difficulties of 
        then getting precisely the same as SELECT * INTO OUTFILE 

        Note that use of `store_result` rather than `use_result` means 
        that all rows are in memory at once.

        NB for consistency the CSV ouput by this command MUST MATCH that 
        by _write_outfile

        `_write_csvdirect` is used by **rdumpcat** , this mimics 
        the output from `_write_outfile` (used by **dumpcat**) with 
        the big advantage that it works remotely, with no strong permission 
        requirements

        TODO:

        #. when there is a pre-existing LOCALSEQNO redirect LOCALSEQNO to a temporay file 
           and do a merge...  easiest to instanciate them as AsciiCSV and then merge at that level 

        """
        q = str(select)
        log.debug("_write_csvdirect %s " % q) 

        llconn = self.llconn   
        llconn.query( q )

        lessmemory = True
        if lessmemory:
            log.debug("using `--LESSMEMORY` option : less memory expensive but more network expensive 'use_result'  ")
            result = llconn.use_result()
        else:
            log.debug("using more memory expensive but less network expensive 'store_result' ")
            result = llconn.store_result()

        csvf = CSVFormat( result.describe() )   
        for row in result.fetch_row(maxrows=0, how=0):   ## all rows as tuples
            tf.write( str(csvf) % tuple(row) +"\n" )


    def timestamped_dir(self, *args):
        """
        Timestamping is needed for non-partitioned case 
        """
        bdir = self.determine_basedir(*args)
        odir = os.path.join(bdir, self.opts.timestamp )
        return odir

    def dumplocal___(self, *args, **kwa ):
        """
        :param outdir:  specifies output directory which must be writable by mysql user, it will be created if not existing 

        Rerunning this will do quick checks of  the CSV files, looking at line counts 
        and the first and last line and comparing with expections from DB queries. The quick
        checks are done via commands:

        * `wc` 
        * `head -1` 
        * `tail -1`

        This is not called in the partitioned case.
        """
        where = kwa.pop('where',None)
        if where is None:
            where = self.opts.where

        tables = kwa.pop('tables',None)
        if tables is None:
            tables = self.utables

        odir = self.timestamped_dir(*args)
        if os.path.exists(odir):
            log.info("dumplocal_ timestamped dir %s exists already, skipping dump" % odir ) 
        else:
            log.info("dumplocal_ into timestamped dir %s " % odir ) 
            self._dumplocal_mkdir(odir)
            for table in tables:
                self._dumplocal_schema(odir, table)
                self._dumplocal_table(odir, table, where)
            pass
        pass
        if self.opts.archive or self.opts.archiveforce:
            self.archive(odir)


    def fields(self, table, filter_=None):
        return filter(filter_,map(lambda _:_['Field'],self("describe %(table)s" % locals())))

    def range(self, table, where="1=1", keycount=True):
        sql = self.partitionmgr.qrange(table, where, keycount=keycount)
        return self(sql)[0]

    def minmax(self, table, where="1=1"):
        sql = self.partitionmgr.qminmax(table, where)
        return self(sql)[0]

    def ptables(self):
        """
        :return: list of tables with the key field
        """
        key = self.partitionmgr.key 
        return filter(lambda t:key in self.fields(t), self.utables)


    def tabminmax_csv(self, path):
        kpo = self.partitionmgr.kpo
        head = os.popen("head -1 %(path)s" % locals()).read().strip()
        tail = os.popen("tail -1 %(path)s" % locals()).read().strip()
        table = os.path.basename(path[:-4]) 
        minkey = int(head.split(",")[kpo])
        maxkey = int(tail.split(",")[kpo])
        return table, minkey, maxkey 


    def _wc_csv(self, outdir, keycount=False): 
        """
        :param outdir: partition directory 
        :return: dict of keyed by table name providing csv info, line count,  

        ::

            {'count': 10, 'keycount': 10, 'max': 40, 'min': 31}


        #. getting the keycount in a general manner, 
           ie number of distinct key values would requiring 
           parsing the entire CSV so fake it from head and tail

        ::

            [blyth@belle7 ~]$ wc -l /tmp/pp/tmp_ligs_offline_db_0/1000/1/*.csv
            192000 /tmp/pp/tmp_ligs_offline_db_0/1000/1/DqChannel.csv
            192000 /tmp/pp/tmp_ligs_offline_db_0/1000/1/DqChannelStatus.csv
              1000 /tmp/pp/tmp_ligs_offline_db_0/1000/1/DqChannelStatusVld.csv
              1000 /tmp/pp/tmp_ligs_offline_db_0/1000/1/DqChannelVld.csv
            386000 total
        """
        if not os.path.exists(outdir):
            return None

        pipe = os.popen("wc -l %(outdir)s/*.csv" % locals())
        lines = pipe.readlines()
        rc = pipe.close()
        if rc is None:
            rc = 0 
        rc = os.WEXITSTATUS(rc) 
        log.debug("wc return code %(rc)s for %(outdir)s " % locals() ) 

        if rc != 0:
            log.warn("Problem with csv files in %(outdir)s wc rc %(rc)s " % locals() ) 
            return None

        wc = {}
        for n, path in map(lambda _:_.lstrip().rstrip().split(), lines):
            if path[-4:] != '.csv':continue
            pass
            table, minkey, maxkey = self.tabminmax_csv(path)
            wc[table] = dict(count=int(n),min=minkey,max=maxkey)
            if keycount:
                wc[table]['keycount'] = maxkey - minkey + 1      # this is quick and dirty 
        return wc

    def determine_basedir(self, *args):
        """
        """
        if len(args)>0:
            basedir = args[0]
        else:
            basedir = self.backupdir  # eg /var/dbbackup/dbsrv/belle7.nuu.edu.tw/channelquality_db_belle7
        pass
        if basedir != self.backupdir:
            log.warn("using basedir %s different from standard %s " % (basedir,self.backupdir) ) 
        else:
            log.debug("using basedir %s " % basedir ) 

        return basedir


    def assert_valid_dump(self, pdir, csv, chk ):
         assert csv == chk , ("prior dump %s check fail" % pdir, pformat(csv),pformat(chk))
         assert sorted(csv.keys()) == sorted(chk.keys()), ("tables mismatch", sorted(csv.keys()),  sorted(chk.keys()))
         for table in csv.keys():
             if chk[table].has_key('keycount'):
                 assert chk[table]['keycount'] == csv[table]['keycount'], (table, chk[table]['keycount'], csv[table]['keycount'] )
             assert chk[table]['count'] == csv[table]['count'], (table, chk[table]['count'], csv[table]['count'] )
             assert chk[table]['min'] == csv[table]['min'], (table, chk[table]['min'], csv[table]['min'] )
             assert chk[table]['max'] == csv[table]['max'], (table, chk[table]['min'], csv[table]['max'] )


    def partition_dumpcheck(self, pdir, pwhere, is_last, keycount=False ):
        """
         Checks a partition dump returning flag to signal a dump or not.

        :param pdir:
        :param pwhere:
        :param is_last:
        :param keycount: doing distinct keycount is quite slow, so can skip for pre-existing
        :return: pdump, chk
        """
        log.info("_")
        log.info("partition_dumpcheck partition loop %s " % pwhere)
        if is_last:
            assert pdir[-5:] == "/last"  , "unexpected pdir %s for last partition " % (pdir) 

        ptables = self.ptables()
        pdump = None 

        # DB look at SEQNO ranges
        chk = {}
        for table in ptables:
            chk[table] = self.range(table, pwhere, keycount=keycount)  

        # file system look at csv files 
        log.info("checking prior csv dump %s  --partitionsize %s --partitionrange %s  " % (pdir,self.opts.partitionsize, self.opts.partitionrange))
        csv = self._wc_csv(pdir, keycount=keycount)
        if csv is None:
            if os.path.exists(pdir):
                msg = "Partition directory \"%(pdir)s\" exists but it contains no .csv files, delete the empty directory and rerun to clear this error" % locals()
                log.fatal(msg)
                raise Exception(msg) 

        # compare DB expectations to CSV dumps 
        if csv == chk:
            log.info("partition dump %s unchanged wrt DB check " % pdir )
            pdump = False
        else:
            pdump = True
            log.info("chk %s " % repr(chk))
            log.info("csv %s " % repr(csv))
            if is_last:
                log.info("last partition dump %s and is changed wrt DB : will redump " % pdir )
            elif csv is None:
                log.info("non last partition dump %s csv is None : first dump " % pdir )
            else:
                log.fatal("non last partition dump %s and is changed wrt DB : will fail " % pdir )
                self.assert_valid_dump( pdir, csv, chk)
            pass
        return pdump, chk 

 
    def partition_dumplocal___(self, *args, **kwa):
        """
        """
        pm = self.partitionmgr 
        pm.basedir = self.determine_basedir(*args)
        if not os.path.exists(pm.basedir):
            os.makedirs(pm.basedir)   

        svy = pm.survey(self)
        ptables = self.ptables()
        assert len(ptables), ptables

        # write all table schema into single schema dir rather than repeating for every partition
        pdir = pm.dir("_")
        if os.path.exists(pdir):
            log.info("schema dir %s exists already" % pdir )
        else:
            log.info("creating and populating schema dir %s " % pdir )
            self._dumplocal_mkdir(pdir)
            for table in ptables:
                self._dumplocal_schema(pdir, table)
            pass
        pass
        if self.opts.archive or self.opts.archiveforce:
            self.archive(pdir)
 
        for p in pm.parts:
            pdir = pm.dir(p)
            pmin = pm.min(p)     
            pmax = pm.max(p)
            pwhere = pm.where(p)
            is_last = pm.is_last(p)   # last partition expected to be incomplete
            keycount = not os.path.exists(pdir)     # only keycount for new partitions
            pdump, chk = self.partition_dumpcheck( pdir, pwhere , is_last, keycount=keycount )
            if pdump:
                log.info("dumplocal partition %s %s %s:%s --partitionsize %s --partitionrange %s " % (p,self.opts.partitionkey,pmin,pmax,self.opts.partitionsize, self.opts.partitionrange)) 
                self._dumplocal_mkdir(pdir)
                for table in ptables:
                    if is_last:
                        log.warn("skipping completeness checks for last partition %s into %s " % (p, pdir))
                        self._dumplocal_table(pdir, table, pwhere)
                    elif chk[table]['min'] == pmin  and chk[table]['max'] == pmax:
                        self._dumplocal_table(pdir, table, pwhere)
                    else:
                        log.warn(" table %s check min  %s pmin %s " % ( table, chk[table]['min'], pmin ))
                        log.warn(" table %s check max  %s pmax %s " % ( table, chk[table]['max'], pmax ))
                        log.warn("skipping dump as check ahead query indicates incomplete partition %s " % repr(check)) 
                    pass
                zdump, zchk  = self.partition_dumpcheck( pdir, pwhere, is_last, keycount=True )
                assert zdump == False, ("post dump dumpcheck signals need to dump", pformat(zchk)) 
            pass
            if self.opts.archive: 
                force = self.opts.archiveforce or pdump
                self.archive(pdir, force=force)
        pass

    def extract(self, dir, base):
        """
        :param dir: directory to be created by extraction 
        :param base: 
        """
        pass
        tgzp, rpath = self.archivepath(dir, base) 
        log.info("extract dir %s tgzp %s rpath %s " % (dir, tgzp, rpath))
        tgz = Tar(tgzp, toplevelname=rpath )
        tgz.examine()
        ls = tgz.list(tgz.members)
        tgz.digest(verify=True)
        tgz.extract( containerdir=base , toplevelname=rpath, dryrun=False )


    def archivepath(self, dir , base=None):
        """
        :param dir: directory to be archived or extracted into 
        :return: path to archive tarball, dir path relative to base
        """ 
        if base is None:
            base = self.backupdir  # eg /var/dbbackup/dbsrv/belle7.nuu.edu.tw/channelquality_db_belle7
        assert dir[0:len(base)] == base, (dir, base, "dir %s should be within base %s " % (dir,base))
        rpath = dir[len(base)+1:]
        elems = rpath.split("/")
        name = "_".join(elems)
        top = elems[0]         # typically this is chunk size
        tgzp = os.path.join(base,"archive", top, "%s.tar.gz" % name )   
        log.debug("archivepath,  dir %s base %s rpath %s name %s tgzp %s  " % ( dir, base, rpath, name, tgzp ) )
        return tgzp, rpath 


    def archive(self, dir , force=False):
        """
        :param dir: directory the contents of which should be archived 

        As a partition corresponds to a certain SEQNO range, it never changes
        so there is no need for a datestring in the path.

        The configured backupfold needs to be created before using the archive `-a` option with::

            [blyth@belle7 DybPython]$ sudo mkdir /var/dbbackup/dbsrv
            [blyth@belle7 DybPython]$ sudo chown -R blyth.blyth /var/dbbackup/dbsrv/

        """
        base = self.backupdir  # eg /var/dbbackup/dbsrv/belle7.nuu.edu.tw/channelquality_db_belle7
        tgzp, rpath = self.archivepath(dir) 
        tgzd = os.path.dirname(tgzp)
        if not os.path.exists(tgzd):
            os.makedirs(tgzd)
        pass

        if self.opts.transfer:
            cfg = self.opts.transfercfg.split(":")
            if len(cfg) == 2:
                kwa = dict(remotenode=cfg[0], remoteprefix=cfg[1])
            elif len(cfg) == 1:
                kwa = dict(remotenode=cfg[0])
            else:
                assert 0, "unexpected transfercfg %s " % repr(cfg)
            log.debug("using transfercfg : %s " % repr(kwa))
        else:
            kwa = {} 
       
        tgz = Tar(tgzp, toplevelname=rpath, **kwa)
        if os.path.exists(tgzp) and not force:
            log.info("archive already exists %s rerun with `-A/--archiveforce` option to recreate " % tgzp )
        else:
            log.info("creating archive %s for %s " % (tgzp,rpath) )
            tgz.archive( base )  # argument specifies the root directory of the archive
        pass
        tgz.examine()
        #log.info("\n".join(tgz.names))
        ls = tgz.list(tgz.members)
        log.debug("\n"+ls)
        du = os.popen("du -hs %(tgzp)s" % locals()).read().strip()
        log.info(du)
        tgz.digest()

        if self.opts.transfer:
            tgz.transfer()


    def partition_loadlocal___(self, *args, **kwa ):
        """
        #. look into putting the partitions back togther again, in partitioned load local
        #. read file system tealeaves wrt the partitioning

        #. factor off the checking 
        #. need to work out which partitions are new and just load those         
        """
        basedir = self.determine_basedir(*args)
        log.info("basedir %s " % basedir )
        pm = self.partitionmgr 
        pm.basedir = basedir

        if self.opts.extract:
            chunks = pm.archived_chunks()
            log.info("archived_chunks %s " % repr(chunks) )
            pm.assign_parts( chunks )
        else:
            chunks = pm.available_chunks()  # only works after extraction
            log.info("available_chunks %s " % repr(chunks) )

        _dir = pm.dir("_")
        if not os.path.exists(_dir):
            log.info("schema dir %s does not exists" % _dir )
            if self.opts.extract:
                self.extract(_dir, basedir)
            pass
        pass
        assert os.path.exists(_dir), _dir 
        #_tables = map(lambda _:_[:-7],filter(lambda _:_[-7:] == '.schema',os.listdir(_dir)))

        for p in pm.parts:
            pdir = pm.dir(p)
            pnam = os.path.basename(pdir)
            pmin = pm.min(p) 
            pmax = pm.max(p) 
            pwhere = pm.where(p) 
            #log.debug("p %s pdir %s pmin %s pmax %s pwhere %s " % (p,pdir,pmin,pmax,pwhere))
            if not os.path.exists(pdir):
                log.debug("partition_loadlocal___ partition dir %s does not exist " % pdir )
                if self.opts.extract:
                    self.extract(pdir, basedir)

            if not os.path.exists(pdir):
                log.debug("partition_loadlocal___ partition dir %s STILL does not exist " % pdir ) 
            else:
                log.info("partition_loadlocal___ loading %s " % pdir ) 
                ltables = self.loadlocal_dir( pdir )
                log.debug("ltables %s " % str(ltables))

                # check that the loaded partition yields the expected key range and count
                if not self.opts.nocheck:
                    check = {}
                    for table in ltables:
                        check[table] = self.minmax(table, pwhere)
                        assert check[table]['min'] == pmin, (table,"min",check[table]['min'],pmin)
                        if pnam != 'last':
                            assert check[table]['max'] == pmax, (table,"max",check[table]['max'],pmax)
                        # keycount check is too slow
                        #assert check[table]['keycount'] == pm.size, (table,"keycount",check[table]['keycount'],pm.size)
                        pass 
                    log.debug(pformat(check))
                pass


    def loadlocal_dir(self, dir ):
        """
        """
        # if a sibling dir named "_" exists use that as a source of schema files, otherwise get directly
        _dir = os.path.join(os.path.dirname(dir),"_")
        if not os.path.exists(_dir):
            _dir = dir 

        def _replace_ignore(table):
            """
            hmm DBI specificity slipping in
            """
            if table == "LOCALSEQNO": 
                return "REPLACE" 
            else:
                return "IGNORE"    

        utables = self.utables

        log.info("utables %s " % repr(utables))

        ltables = [] 
        for name in filter(lambda _:os.path.isfile(os.path.join(dir,_)) and _[-4:] == '.csv' , os.listdir(dir)):
            path = os.path.join(dir, name)
            table, ext = os.path.splitext(name)
            if not table in utables:
                schema = os.path.join(_dir, "%s.schema" % table )
                log.warn("creating table %s from schema file %s " % (table, schema))
                assert os.path.exists(schema), schema
                self(open(schema,"r").read())
            pass
            ltables.append(table)

            ttable, csvminkey, csvmaxkey = self.tabminmax_csv(path)
            assert csvminkey < csvmaxkey , (csvminkey,csvmaxkey)
            assert ttable == table, (ttable, table)
            mm = self.minmax(table)            
            log.debug("csvminkey %s csvmaxkey %s " % (csvminkey, csvmaxkey))
            log.debug(" mmmin %(min)s  mmmax %(max)s " % mm )
            if csvmaxkey <= mm['max']:
                log.info("SKIP: as already loaded csv keys minmax %s %s from %s " % (csvminkey,csvmaxkey,path ))
            else:
                ll = LoadDataLocalInfile(infile=path, table=table, ignorelines=0, replace_ignore=_replace_ignore(table) )
                self(str(ll))        
                log.info("loadlocal ingesting %s took %s seconds " % (path, "%5.2f" % self.lastseconds ) ) 
        pass 
        return ltables
 

    def loadlocal___(self, *args, **kwa ):
        """
        :param outdir:  specifies directory containing normal or partitioned dump of CSV files
        """
        odir = self.timestamped_dir(*args)
        return self.loadlocal_dir(odir)
  


class tdict(dict):
    __str__ = lambda self:self.tmpl % self

class IntoOutfile(tdict):
    tmpl = "select * from %(table)s where %(where)s into outfile '%(outfile)s' fields terminated by ',' optionally enclosed by '\"' "

class RemoteIntoOutfile(tdict):
    tmpl = "select * from %(table)s where %(where)s "

class LoadDataLocalInfile(tdict):
    tmpl = "LOAD DATA LOCAL INFILE '%(infile)s' %(replace_ignore)s INTO TABLE %(table)s FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' IGNORE %(ignorelines)s LINES " 



class PartitionMgr(object):
    int_ptn=re.compile("^\d*$")
    def __init__(self, opts, basedir=None):
        """
        #. for example 1000,0,3 for three chunks of 1000 starting at 1 
        """
        self.basedir = basedir  
        pass
        size = int(opts.partitionsize)  
        pass
        if opts.partitionrange is None:
            parts = None
        else:
            a, n = map(int,opts.partitionrange.split(","))  
            parts = range(a,n)
        pass  
        key, kpo = opts.partitionkey.split(",")
        and_where = opts.where  
        pass
        self.and_where = and_where

        if opts.partitionlast is None:
            plast = None
        else:             
            plast = int(opts.partitionlast) 
        pass
        self.plast = plast
        pass
        self.key = key       # key string eg SEQNO
        self.kpo = int(kpo)  # key position in csv
        pass
        self.size = size
        self.parts = parts


    def survey(self, db ):
        """
        Need to identify the last partition after which not 
        all of the tables will have full partitions.

        For sensible partitioning should restrict to tables 
        whose SEQNO stride along roughly together. 

        """
        size = self.size
        log.info("PartitionMgr survey size %s " % size  )
        svy = {}
        mlp = 0
        
        for table in db.ptables():
            mm = db(self.qminmax(table,where="TRUE"))[0]
            last = mm['max']/size
            svy[table] = dict(mm, last=last )
        pass
        svy['COMMON'] = dict(
            min=max(map(lambda _:_['min'],svy.values())),  
            max=min(map(lambda _:_['max'],svy.values())),
            last=min(map(lambda _:_['last'],svy.values()))
          )  

        log.info("\n"+pformat(svy)) 
        autoplast = svy['COMMON']['last']
        log.info("auto survey determines last partition index as %s " % autoplast )
        if self.plast is None:
            self.plast = autoplast  
        else:
            if self.plast != autoplast:
                log.warn("`--partitionlast` option gives %s BUT auto-survey gives %s " % (self.plast, autoplast))  
            pass
        pass
        if self.parts is None:
            parts = range(0,self.plast+1)  # plast is the 0-based index of the last partition, hence the need for  +1
            log.info("auto definition of partition range %s " % repr(parts))
            self.parts = parts 
        else:
            log.info("using partition range from options %s " % repr(self.parts))
        return svy
 
    def assign_parts( self, chunks ):
        """
        :param chunks:
        """
        parts = sorted(chunks)
        plast = parts[-1] + 1
        parts += [plast]

        self.parts = parts
        self.plast = plast
        log.info("assign_parts %s plast %s " % ( self.parts, self.plast ))

    def dir(self, p): 
        """
        """ 
        if self.is_last(p):
            leaf = "last"
        else:
            leaf = p
        pass
        return os.path.join(self.basedir,str(self.size),str(leaf)) 

    def is_last(self, p):
        return p == self.plast

    def qminmax(self, table, where="1=1"):
        key = self.key
        return "select max(%(key)s) as max, min(%(key)s) as min, count(*) as count from %(table)s where %(where)s " % locals()

    def qrange(self, table, where="1=1", keycount=True):
        key = self.key
        if keycount:
            keycount = ", count(distinct(%(key)s)) as keycount " % locals()
        else:
            keycount = ""
        pass
        return "select max(%(key)s) as max, min(%(key)s) as min, count(*) as count  %(keycount)s from %(table)s where %(where)s " % locals()

    def offset(self, p): 
        return self.size*p 
    def min(self, p): 
        return self.size*p + 1
    def max(self, p): 
        return self.size*(p+1)
    def where(self, p):
        n = len(self.parts) 
        size = self.size 
        pmin = self.min(p) 
        pmax = self.max(p) 
        pkey = self.key
        p1 = p + 1
        if self.and_where is None:
            and_where = ""
        else:
            and_where = " and %s " % self.and_where 

        return "/* [%(p)s] %(size)s-partition %(p1)-3s/%(n)s  */ %(pkey)s >= %(pmin)s and %(pkey)s <= %(pmax)s %(and_where)s " % locals()

    def int_dirname(self, dir):
        """
        :param dir:
        :return: list of integers corresponding to subfolder of *dir* with integer names, in ascending order 
        """ 
        return sorted(map(int,filter(lambda _:self.int_ptn.match(_), os.listdir(dir))))

    def has_partitionset(self):
        """
        Checks if there is a partitionset folder for the configured partition size.
        ie looks for the 10000 folder, in default case.
        """
        psizes = self.int_dirname(self.basedir)
        if not self.size in psizes:
            msg = "partition dump at %s does not have partitions of the configured size %s : %s " % (self.basedir, self.size, psizes)
            log.warn(msg)
            return False 
        return True 

    def archived_chunks(self):
        """
        :return: list of integers in ascending order corresponding to partition archives found
        """
        zdir  = os.path.join(self.basedir, "archive", str(self.size))
        ptn = re.compile("^(?P<size>\d*)_(?P<part>\d*).tar.gz.dna$")
        chunks = []
        for path in os.listdir(zdir):
            name = os.path.basename(path)
            match = ptn.match(name)
            if match:
                d = match.groupdict()
                chunks.append(int(d['part']))
            pass 
        return sorted(chunks)

    def available_chunks(self): 
        """
        :return: list of intergers in ascending order corresponding to the partition folders found

        Looks for the partition folder, named according to the partition size eg "basedir/10000" 
        and determines the chunks by looking at the integer named partition subfolders "basedir/10000/0" 
        """
        if not self.has_partitionset():
            log.warn("no partitionset for partition size %s" % self.size)
            return []  
        pfold = os.path.join( self.basedir, str(self.size) ) 
        chunks = self.int_dirname(pfold)
        log.info("partition folder %s for configured size %s has chunks %s " % (pfold, self.size, repr(chunks)) )
        return chunks


class MyCnf(dict):
    def __init__(self, path = "~/.my.cnf", prime={}): 
        """
        :param path: to config file
        :param prime: initial dict to prime the config, eg for current date
        """
        cfp = ConfigParser(prime)
        paths = cfp.read( [os.path.expandvars(os.path.expanduser(p)) for p in path.split(":")] )
        log.debug("MyCnf read %s " % repr(paths) )
        self.cfp = cfp 
        self.sections = cfp.sections() 
        self.path = path
        self.paths  = paths

    def section(self, sect, home=None):
        """
        :param sect: name of section in config file
        :return: dict of config parameters from specified section of config file

        When the section is present in the config file simply return
        the parameters. If the section is not present in the file then 
        adopt the \"home\" instanance section and swap in the the database
        from the sect name.

        This allows access to all databases on a server without having 
        corresponding config file sections for all of them.
        """
        if sect in self.sections:
            it = dict(self.cfp.items(sect))
        else:
            if not home is None:
                assert home.sect in self.sections, self.sections
                it = dict(self.cfp.items(home.sect))
                it['database'] = sect
                log.debug("no section %s infer config from home.sect %s section assuming section name = dbname " % (sect, home.sect) )
            else:
                it = None
        pass
        return it

    def mysqldb_pars(self, sect, home=None):
        """
        :param sect: name of section in config file
        :param home: DB instance 
        :return: dict of mysql-python style connection parameters

        Annoyingly mysql-python needs these keys

        `host` host to connect
        `user` user to connect as
        `passwd` password to use
        `db` database to use

        whereas mysql uses slightly different ones

        `host`   
        `user`
        `password`
        `database` 

        Normally can avoid this annoyance using::

            conn = MySQLdb.connect( read_default_group=sect )   

        but when need to impinge `database/db` settings this is not possible.
        """
        my2mp = dict(host="host",user="user",password="passwd",database="db", socket="unix_socket")
        my = self.section(sect, home=home)

        if my is None:
           msg = "missing required section \"%s\" in config file \"%s\" " % ( sect, self.path )
           log.fatal(msg)
           raise Exception(msg)

        mp = {}
        for k in filter(lambda k:k in my2mp,my.keys()):  # key translation, mysql to mysql-python
            mp[my2mp[k]] =  my[k]
        log.debug("translate mysql config %s into mysql-python config %s " % ( dict(my,password="***") , dict(mp,passwd="***") ))
        return mp 





def main():
    """
    Need to support logging on py2.3

    http://docs.python.org/release/2.3.5/lib/module-logging.html
    """
    from optparse import OptionParser
    op = OptionParser(usage=__doc__ + "\n" + DB.docs() )
    now = datetime.now().strftime("%Y%m%d_%H%M")

    tables = "DqChannel,DqChannelVld,DqChannelStatus,DqChannelStatusVld"   # default tables to backup/propagate

    op.add_option("-l", "--loglevel", default="INFO", help="Choose logging level case insensitively eg INFO,DEBUG,WARN. Default %default " )
    op.add_option(      "--logformat", default="%(asctime)s %(name)s %(levelname)-8s %(message)s" , help="Logging format. Default %default " )
    op.add_option("-t", "--tables",   default=tables,   help="Comma delimited list of table names to be included in operations. Default %default " )
    op.add_option(      "--DB_DROP_CREATE", action="store_true",   help="CAUTION: DROPs and CREATES the database specified in the first argument. Default %default " )
    op.add_option("-C", "--noconfirm", action="store_true",   help="Skip interactive confirmation of dangerous operations. Default %default " )
    op.add_option("-K", "--nocheck", action="store_true",   help="Skip some slow checking. Default %default " )
    op.add_option("-w", "--where",   default=None,   help="Where clause applied to selects on all used tables. Default %default " )
    op.add_option(      "--partition",    action="store_true",     help="Use partitioned dumping OR loading, used as an optimization for dealing with very large tables. Default %default " )
    op.add_option(      "--partitionrange",  default=None,    help="Partition range in partition indices corresponding to python `range(a,b)` or None for all. Default %default " )
    op.add_option(      "--partitionsize",   default="10000",   help="Partition size. Default %default " )
    op.add_option(      "--partitionkey",   default="SEQNO,0",     help="Fieldname, field position in CSV on which partition chunking is based. Default %default " )
    op.add_option(      "--partitionlast",   default=None,         help="(DEPRECATED now determined automatically) Used to identify the index of the last incomplete partition, in order to skip completeness checks and save into a \"last\" directory. Default %default " )
    op.add_option(      "--timestamp",      default=now,  help="Non-partitioned dumps and archives are placed inside date stamped folders, to use a prior one the stamp must be specifed. Default %default " )
    op.add_option( "-A","--archiveforce",   action="store_true",     help="Proceed with archiving even if a preexisting archive exists. Default %default " )
    op.add_option( "-a","--archive",     action="store_true",     help="Create archives of the backups ready for offbox transfer. Default %default " )
    op.add_option( "-x","--extract",     action="store_true",     help="Extracts backup folders from archives. Default %default " )
    op.add_option(      "--backupfold",  default="/var/dbbackup/dbsrv",   help="Folder under which backups and archives are kept. Default %default " )
    op.add_option( "-T","--transfer",   action="store_true",     help="Must be used together `--archive` option to transfer tarballs to the remote node configured with `--transfercfg`. Default %default " )
    op.add_option(      "--transfercfg",   default=os.environ.get('DBSRV_REMOTE_NODE','S:/data'), help="Configure archive transfers with the ssh node name and destination prefix eg \"S:/data\" that comes ahead of the backupfold. Default %default " )
    op.add_option( "-s","--home",           default="loopback",     help="Name of .my.cnf section of information_schema DB on server regarded as home. Should normally be \"loopback\". Default %default " )
    opts, args = op.parse_args()
    level = getattr(logging, opts.loglevel.upper())
    try: 
        logging.basicConfig(format=opts.logformat,level=level)
    except TypeError:
        hdlr = logging.StreamHandler()
        formatter = logging.Formatter(opts.logformat)
        hdlr.setFormatter(formatter)
        log.addHandler(hdlr)
        log.setLevel(level)
    pass
    assert len(args)>1, "need 2 or more arguments %s " % repr(args)

    log.info("sys.version_info %s " % repr(sys.version_info)) 
    log.info("got %s args : %s " % (len(args), repr(args))) 

    if opts.transfer:
        sshenv()   # pick up ssh environment for use from spartan environments such as the cron commandline  

    home = DB(opts.home, opts=opts)
 
    arg = args[0]
    if arg.find(",") > -1:
        log.debug("comma delimited argument %s " % arg )
        names = arg.split(",")
    elif arg.find("\\") > -1 or arg.find("\*") > -1:
        log.debug("regexp argument %s " % arg )
        ptn = re.compile(arg)
        names = filter(lambda dbname:ptn.match(dbname), home.databases)
    else:
        log.debug("plain single DB argument %s " % arg )
        names = [arg]
    pass
    log.debug("arg %s names %s " % ( arg, repr(names)))
    for name in names: 
        if opts.DB_DROP_CREATE:    
            home.database_drop_create(name)
        pass
        db = DB(name, opts=opts, home=home)
        db.dispatch( *args[1:] ) 

if __name__=='__main__':
    db = main()