#!/usr/bin/python

"""Implements datacycle reading, parsing and storing"""

import atexit
import getopt
import io
import logging
import operator
import os
import pwd
import re
import sqlite3
import signal
import stat
import subprocess
import sys
import time

from functools import wraps

__version__ = '0.3.4'


def terminate_rivals(db_path):
    try:
        inode_str = str(os.stat(db_path).st_ino)
        self_pid_str = str(os.getpid())
        with open('/proc/locks') as f:
            locks = f.read().strip().splitlines()
        for lock in locks:
            if inode_str not in lock:
                continue
            if self_pid_str in lock:
                continue
            items = lock.split(None, 5)     # items[4] is locking process PID
            logging.warn('Terminating rival with PID %s...' % (items[4],))
            os.kill(int(items[4]), signal.SIGTERM)
    except Exception, e:
        logging.error('Could not terminate a rival: %s' % (e,))
        return


def enable_wal(dbfile, enable=True):
    logging.debug('Enabling WAL mode')
    if os.path.exists(dbfile):
        query = 'PRAGMA journal_mode={0}'
        if enable:
            query = query.format('WAL')
        else:
            query = query.format('DELETE')

        with sqlite3.connect(dbfile) as conn:
            table = conn.execute(query)

            for row in table:
                for cell in row:
                    logging.debug('Current table mode: {0}'.format(cell))


def retry(fn):
    @wraps(fn)
    def wrapper(self, *args, **kw):
        for _ in range(self._retries):
            try:
                return fn(self, *args, **kw)
            except sqlite3.OperationalError as e:
                logging.warn("DB error occurred: %s -> %s. Retrying..." % (
                    fn.__name__, e))
                time.sleep(self._timeout)
        logging.error("DB error occurred: %s" % (fn.__name__,))
        if self._terminate_rivals:
            terminate_rivals(self._db_path)
    return wrapper


class DB(object):

    tablename = 'log'

    def __init__(self, db_path, timeout=0.1, retries=2, check=True,
                 terminate_rivals=False):
        self._db_path = db_path
        self._timeout = timeout
        self._retries = retries
        self._check = check
        self._terminate_rivals = terminate_rivals

    def __enter__(self):
        if self._check:
            self._create_db_if_missing()
        self._conn = sqlite3.connect(self._db_path)
        self._cur = self._conn.cursor()
        if self._check:
            self._create_table_if_missing()
        return self

    def _create_db_if_missing(self):
        logging.debug("Checking DB...")
        try:
            if not os.path.exists(self._db_path):
                db_dir = os.path.dirname(self._db_path)
                if not os.path.exists(db_dir):
                    os.makedirs(db_dir)
                open(self._db_path, 'a').close()
                os.chmod(self._db_path, stat.S_IRUSR | stat.S_IWUSR)
            enable_wal(self._db_path)
        except (IOError, OSError):
            return

    def _create_table_if_missing(self):
        logging.debug("Checking DB table...")
        query = "SELECT name FROM sqlite_master WHERE type='table'"
        result = self._exec(query)
        if not result:
            return
        tables = [i[0] for i in result.fetchall()]

        if self.tablename in tables:
            return False

        q1 = ("CREATE TABLE %s (uid INTEGER NOT NULL, timestamp INTEGER(4) "
              "NOT NULL, filename TEXT NOT NULL)" % (self.tablename,))

        q2 = ("CREATE INDEX idx_%s_uid on %s (uid)" % (
                self.tablename, self.tablename))

        q3 = ("CREATE INDEX idx_%s_timestamp on %s (timestamp)" % (
                self.tablename, self.tablename))

        for q in q1, q2, q3:
            self._exec(q)
        return True

    @retry
    def put_all(self, data):
        query = "INSERT INTO %s (uid, timestamp, filename) VALUES (?,?,?)" % (
            self.tablename,)
        return self._cur.executemany(query, data)

    def put(self, data):
        query = "INSERT INTO %s (uid, timestamp, filename) VALUES (?,?,?)" % (
            self.tablename,)
        self._exec(query, data)

    def delete_old(self, offset=None):
        if offset is None:
            offset = int(time.time()) - 86400        # 1 day
        query = "DELETE FROM %s WHERE timestamp < ?" % (self.tablename,)
        self._exec(query, (offset,))

    @retry
    def _exec(self, query, data=()):
        return self._cur.execute(query, data)

    @retry
    def _commit(self):
        return self._conn.commit()

    def __exit__(self, exc_type, exc_value, traceback):
        self._commit()
        self._conn.close()


class NullHandler(logging.Handler):
    def emit(self, record):
        pass


class Daemon(object):
    """Implements general daemon process"""
    pidfile = '/tmp/daemon.pid'

    def __init__(self):
        signal.signal(signal.SIGTERM, self._sigterm_handler)
        signal.signal(signal.SIGHUP, self._sighup_handler)

    def start(self):
        pid = self._get_pid()
        if pid is not None:
            message = "pidfile %s exists. Is daemon already running?\n"
            raise SystemExit(message % (self.pidfile,))
        self._start()

    def stop(self):
        pid = self._get_pid()
        if pid is None:
            message = "pidfile %s does not exist. Is daemon stopped?\n"
            sys.stderr.write(message % (self.pidfile,))
            return
        os.kill(pid, signal.SIGTERM)

    def restart(self):
        self.stop()
        self.start()

    def _start(self):
        """
        Subroutine to be redefined by successor
        """
        pass

    def _stop(self):
        """
        Subroutine to be redefined by successor
        """
        pass

    def _daemonize(self):
        if os.path.exists(self.pidfile):
            raise RuntimeError('Already running')
        self._do_fork(1)
        os.chdir('/')
        os.umask(0)
        os.setsid()
        self._do_fork(2)
        sys.stdout.flush()
        sys.stderr.flush()
        with open('/dev/null', 'r', 0) as f:
            os.dup2(f.fileno(), sys.stdin.fileno())
        with open('/dev/null', 'a', 0) as f:
            os.dup2(f.fileno(), sys.stdout.fileno())
        with open('/dev/null', 'a', 0) as f:
            os.dup2(f.fileno(), sys.stderr.fileno())
        self._makepid()
        atexit.register(self._delpid)

    def _sigterm_handler(self, signo, frame):
        self._stop()
        self._delpid()

    def _sighup_handler(self, signo, frame):
        """
        Subroutine to be redefined by successor
        """
        logging.info("Got SIGHUP. Reloading configuration")

    @staticmethod
    def _do_fork(num):
        try:
            if os.fork() > 0:
                raise SystemExit(0)
        except OSError:
            raise RuntimeError('fork #%s failed' % num)

    def _get_pid(self):
        pid = None
        try:
            with open(self.pidfile) as f:
                pid = f.read().strip()
            if pid is not None and pid.isdigit():
                pid = int(pid)
        except (OSError, IOError):
            pass
        return pid

    def _delpid(self):
        try:
            os.unlink(self.pidfile)
        except OSError:
            return

    def _makepid(self):
        mode = os.O_CREAT | os.O_EXCL | os.O_WRONLY
        try:
            fd = os.open(self.pidfile, mode, 0644)
        except OSError:
            sys.exit(1)

        with os.fdopen(fd, 'w') as f:
            f.write(str(os.getpid()))


class LogReader(Daemon):

    _pwd_db = '/etc/passwd'
    buffer_size = 1048576   # 1MB

    def __init__(self, **config):
        super(LogReader, self).__init__()
        logging.getLogger(__name__).addHandler(NullHandler)
        for k, v in config.items():
            super(LogReader, self).__setattr__(k, v)
        self._is_running = False
        self._buff = []
        self._check = True
        self._users = {}
        self._pwd_mtime = 0
        self._path_patt = re.compile(r'/*?(?P<start>[^/]+)(?P<rest>.*)')
        self._update_users_cache_if_needed()

    def _update_users_cache_if_needed(self):
        _pwd_mtime = os.stat(self._pwd_db).st_mtime
        if _pwd_mtime > self._pwd_mtime or not self._users:
            for u in pwd.getpwall():
                self._users[u.pw_name] = u
            self._pwd_mtime = _pwd_mtime

    def _resolve_abs_path(self, path):
        m = self._path_patt.match(path)
        if not m:
            return path
        start, rest = m.group('start', 'rest')
        if os.path.exists('/' + start):
            return path
        if start in self._users:
            return self._users[start].pw_dir + rest
        return path

    def _run(self, events_path, flush_path):
        try:
            prepare_datacycle(self.minimal_event_uid, self.max_events)
            while self._is_running:
                self._get(events_path, flush_path)
                time.sleep(self.polling_interval)
        finally:
            write_int_to_file(
                '/proc/sys/fs/datacycle/user_ro_mode',
                self.user_ro_mode_min_uid)

    def _ext_read(self, events_path):
        cmd = ['cat', events_path]
        p = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=open(os.devnull, 'a'))
        sout, serr = p.communicate()
        if p.returncode != 0:
            logging.error("An error occurred while processing %s" % (cmd,))
            return
        return sout

    def _inn_read(self, events_path):
        buff = bytearray(self.buffer_size)
        with io.open(events_path, 'rb', buffering=0) as r:
            r.readinto(buff)
        return buff


    def _get(self, events_path, flush_path, out=False):
        last_eid = '-1'         # last event id
        timestamp = int(time.time())
        self._update_users_cache_if_needed()

        _read = self._inn_read if self.use_inner_reader else self._ext_read

        num_of_reads = 4

        while num_of_reads > 0:

            sout = _read(events_path)
            if not sout:
                break

            m, start, stop, first_found, iter_valid = None, 0, 0, False, True

            itr = re.finditer(b'(?P<hdr>\d+:\d+:\d+:)', sout, re.DOTALL)

            while iter_valid:

                try:
                    m = next(itr)
                except StopIteration:
                    iter_valid = False

                if not first_found:             # the first match. As we only
                    first_found = True          # can get path while looking
                    if m is not None:           # behind, we set flag to say we
                        start, stop = m.span('hdr')
                    continue                    # can start from next iteration

                if iter_valid:
                    line = sout[start:m.start('hdr')]
                else:
                    line = sout[start:]

                if out:
                    print line

                logging.debug(line)

                try:
                    last_eid, tid, uid, fpath = line.decode('utf8').split(':', 3)

                    if not iter_valid:      # strip zeroes that remain in buffer
                        fpath = fpath.rstrip('\0')

                    if fpath[-1] == '\n':   # if we have double newline we need
                        fpath = fpath[:-1]  # to remove the last one only

                    # strip '//deleted' suffix from path: it has no relation
                    # to a file path as such
                    if fpath.endswith('//deleted'):
                        fpath = fpath[:-9]

                    fpath = self._resolve_abs_path(fpath)

                    if self.include and not any(
                            i for i in self.include if fpath.startswith(i)):
                        if m is not None:
                            start, stop = m.span('hdr')
                        continue

                    if (self.exclude and
                            any(i for i in self.exclude if i in fpath)):
                        if m is not None:
                            start, stop = m.span('hdr')
                        continue

                    if self.event_types and tid not in self.event_types:
                        if m is not None:
                            start, stop = m.span('hdr')
                        continue

                    self._buff.append((str(uid), timestamp, fpath))

                except (ValueError, UnicodeDecodeError):
                    pass

                if m is not None:
                    start, stop = m.span('hdr')

            if last_eid != '-1':
                with open(flush_path, 'w') as f:
                    f.write(last_eid)

            num_of_reads -= 1

        if len(self._buff) == 0:
            return

        with DB(self.dbfile, check=self._check,
                terminate_rivals=self.terminate_rivals) as db:
            rv = db.put_all(self._buff)
            if self.time_to_keep:       # Neither None nor zero
                db.delete_old(timestamp - self.time_to_keep * 86400)
        if rv:
            self._buff = []
            self._check = False
        else:
            logging.warn(
                "Could not record data to DB, %d entries stay in buffer" % (
                    len(self._buff),))

    def _start(self):
        events_path = '/proc/sys/fs/datacycle/events'
        flush_path = '/proc/sys/fs/datacycle/flush'

        for path in events_path, flush_path:
            if not os.path.exists(path):
                raise SystemExit("%s does not exist. Exit" % (path,))

        if not self.no_daemon:
            self._daemonize()
        self._is_running = True
        self._run(events_path, flush_path)

    def _stop(self):
        self._is_running = False

    def get(self):
        events_path = '/proc/sys/fs/datacycle/events'
        flush_path = '/proc/sys/fs/datacycle/flush'
        self._get(events_path, flush_path, out=True)


def usage():
    d_pid = '/var/run/cloudlinux-file-change-collector.pid'
    d_log = '/var/log/cloudlinux-file-change-collector.log'
    print ''
    print 'Usage: ' + sys.argv[0] + ' [OPTIONS] ACTION'
    print ''
    print 'Actions:'
    print ''
    print 'start   : start the service'
    print 'stop    : stop the service'
    print 'restart : restart the service'
    print 'get     : read current events contents without starting service'
    print 'No action implies "start"'
    print ''
    print 'Options:'
    print ''
    print ' -h | --help      : shows this help'
    print ' -c | --config    : config file'
    print ' -p | --pidfile   : pid file (by default %s)' % (d_pid,)
    print ' -l | --logfile   : log file (by default %s)' % (d_log,)
    print ' -D | --debug     : raise log verbosity to debug level'
    print ' -n | --no-daemon : run in non-daemon mode'
    print ''


def set_logger(config):
    level = logging.INFO if config['loglevel'] == 'info' else logging.DEBUG
    logging.basicConfig(
        filename=config['logfile'],
        level=level,
        format='%(asctime)s :%(levelname)s %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S')

    if config['no_daemon']:
        hdlr = logging.StreamHandler()
        hdlr.setLevel(level)
        fmt = logging.Formatter(fmt='%(levelname)s: %(message)s')
        hdlr.setFormatter(fmt)
        logging.getLogger().addHandler(hdlr)


def resolve_events(events):
    """
    Splits events to tokens and maps'em to code digits
    :param events: str -> comma separated list of events
    :return: set -> set of digits as strings, i. e. set(['1', '3'])
    """
    filtered_events = set()
    events_map = {
        'file_created': '0',
        'dir_created': '1',
        'dir_deleted': '2',
        'file_deleted': '3',
        'symlink_created': '4',
        'hardlink_created': '5',
        'moved': '6',
        'attrib_changed': '10',
        'owner_changed': '11',
        'file_modified': '12'}
    events = events.strip(',')      # remove dandgling commas if any
    if events == '' or events == 'all':
        return filtered_events      # Nothing to filter. We'll process all
    for key in [i.strip() for i in events.split(',')]:
        if key in events_map:
            filtered_events.add(events_map[key])
    return filtered_events


def get_config(config):
    if 'conffile' not in config:
        return
    try:
        with open(config['conffile']) as f:
            for line in f:
                if line.startswith('#') or '=' not in line:
                    continue
                try:
                    key, value = [i.strip() for i in line.split('=', 1)]
                    if key == 'database_path':
                        config['dbfile'] = value
                    elif key == 'logfile':
                        config['logfile'] = value
                    elif key == 'include' and value != '':
                        config['include'].add(value)
                    elif key == 'exclude'and value != '':
                        config['exclude'].add(value)
                    elif key == 'polling_interval':
                        if value.isdigit():
                            config['polling_interval'] = int(value)
                    elif key == 'time_to_keep':
                        if value.isdigit():
                            config['time_to_keep'] = int(value)
                    elif key == 'user_ro_mode_min_uid':
                        config['user_ro_mode_min_uid'] = int(value)
                    elif key == 'minimal_event_uid':
                        config['minimal_event_uid'] = int(value)
                    elif key == 'event_types':
                        config['event_types'] = resolve_events(value)
                    elif key == 'terminate_rivals' and value == 'yes':
                        config['terminate_rivals'] = True
                    elif key == 'use_inner_reader' and value == 'yes':
                        config['use_inner_reader'] = True
                    elif key == 'max_events':
                        config['max_events'] = int(value)
                except ValueError:
                    continue
    except (IOError, OSError):
        return


def datacycle_running():
    """
    Endurance provide their datacycle collecting utility named 'datacycle.pl'.
    Here a check for running datacycle.pl is made (to avoid running our process
    at the same time).
    """
    cmd = ['ps', 'aux']
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                         stderr=open(os.devnull, 'a'))
    out, err = p.communicate()
    if p.returncode != 0:
        logging.error("An error occurred while processing %s" % (cmd,))
        sys.exit(1)
    for line in out.splitlines():
        if 'datacycle.pl' in line:
            return True
    return False


def write_int_to_file(path, value):
    try:
        checked = '%d\n' % value    # Raise if no integer before opening file
        with open(path, 'w') as f:
            f.write(checked)
    except (IOError, OSError, TypeError):
        return False
    return True


def prepare_datacycle(minimal_event_uid, max_events):
    if datacycle_running():
        raise SystemExit("'datacycle.pl' is running. "
                         "Running at the same time is useless.")

    file_values = [
        # enable datacycling
        ('/proc/sys/fs/datacycle/enable', 1),
        # Restore users ability to write to their homedirs
        ('/proc/sys/fs/datacycle/user_ro_mode', -1),
        # Set minimal user event UID
        ('/proc/sys/fs/datacycle/min_event_uid', minimal_event_uid),
        # Set upper limit for events buffer
        ('/proc/sys/fs/datacycle/max_events', max_events)]

    for path, value in file_values:
        rv = write_int_to_file(path, value)
        if not rv:
            logging.error("Count not write %s to %s" % (value, path))


def main():
    action = 'start'
    config = {}
    config['conffile'] = '/etc/sysconfig/cloudlinux-fchange'
    config['dbfile'] = '/var/lve/cloudlinux-fchange.db'
    config['pidfile'] = '/var/run/cloudlinux-file-change-collector.pid'
    config['logfile'] = '/var/log/cloudlinux-file-change-collector.log'
    config['loglevel'] = 'info'
    config['polling_interval'] = 3
    config['exclude'] = set()
    config['include'] = set()
    config['event_types'] = set()
    config['time_to_keep'] = None
    config['no_daemon'] = False
    config['user_ro_mode_min_uid'] = -1
    config['minimal_event_uid'] = 500
    config['terminate_rivals'] = False
    config['use_inner_reader'] = False
    config['max_events'] = 65536
    try:
        opts, args = getopt.getopt(
            sys.argv[1:],
            'hDc:p:l:n',
            ['help', 'debug', 'config=', 'pidfile=', 'logfile=', 'no-daemon'])
    except getopt.GetoptError, e:
        raise SystemExit(str(e))

    for o, a in opts:
        if o in ['-h', '--help']:
            usage()
            sys.exit(0)
        elif o in ['-c', '--config']:
            config['conffile'] = a
        elif o in ['-p', '--pidfile']:
            config['pidfile'] = a
        elif o in ['-l', '--logfile']:
            config['logfile'] = a
        elif o in ['-D', '--debug']:
            config['loglevel'] = 'debug'
        elif o in ['-n', '--no-daemon']:
            config['no_daemon'] = True

    if len(args) == 1:
        action = args[0]

    get_config(config)
    set_logger(config)

    r = LogReader(**config)

    try:
        operator.methodcaller(action)(r)
    except AttributeError, e:
        m = re.search(r"""attribute\s+(?P<action>'[^']+')""", str(e))
        prefix = "Unknown action"
        raise SystemExit(
            "%s: %s" % (prefix, m.group('action')) if m else prefix)


if __name__ == '__main__':
    main()
