dig around in internal logs account

internal_logs.py

#!/usr/bin/env python
import argparse
import logging
import sys
import thread
import threading
from Queue import Queue

from datetime import datetime, timedelta

from swift.common.internal_client import InternalClient
from swift.container.sync import ic_conf_body
from swift.common.wsgi import ConfigString

# fix monkey-patch lp bug #1380815
logging.threading = threading
logging.thread = thread
logging._lock = threading.RLock()

TODAY = datetime(*datetime.today().timetuple()[:3])
YESTERDAY = TODAY - timedelta(days=1)
DATE_FORMAT = '%Y/%m/%d'


def parse_date(string):
    try:
        d = datetime.strptime(string, DATE_FORMAT)
    except ValueError as e:
        raise argparse.ArgumentTypeError(str(e))
    return datetime(d.year, d.month, d.day)


def add_start_end(parser):
    parser.add_argument('-S', '--start', default=YESTERDAY, type=parse_date,
                        help='YYYY/MM/DD to start search (inclusive)')
    parser.add_argument('-E', '--end', default=TODAY, type=parse_date,
                        help='YYYY/MM/DD to end search (exclusive)')


parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers()

list_nodes_parser = subparsers.add_parser('list-nodes')

list_logs_parser = subparsers.add_parser('list-logs')
add_start_end(list_logs_parser)

grep_logs_parser = subparsers.add_parser('grep-logs')
add_start_end(grep_logs_parser)
parser.add_argument('--workers', help='number of workers', type=int,
                    default=24)
grep_logs_parser.add_argument('-e', '--pattern', help='something to look for',
                              default='')


def mb(bytes_):
    return '%.02f MiB' % (bytes_ / 2.0 ** 20)


def gb(bytes_):
    return '%.02f GiB' % (bytes_ / 2.0 ** 30)


def do_list_nodes(swift, args):
    listing = swift.iter_containers('.logs')
    total = 0
    for item in listing:
        print '%s - %s - %s' % (
            item['name'], item['count'], gb(item['bytes']))
        total += item['bytes']
    print 'Total: %s' % gb(total)


list_nodes_parser.set_defaults(func=do_list_nodes)


def get_aggr_logs(swift, args):
    nodes = swift.iter_containers('.logs')
    aggr_logs = []
    for node in nodes:
        logs = swift.iter_objects('.logs', node['name'])
        for log in logs:
            if log['name'].endswith('ssnoded.log.gz'):
                continue
            date = parse_date(log['name'][:10])
            if args.start <= date < args.end:
                aggr_logs.append((date, node['name'], log))
    return aggr_logs


def extract_date_from_name(name):
    return datetime.stptime()


def do_list_logs(swift, args):
    aggr_logs = get_aggr_logs(swift, args)
    total = 0
    for date, node, log in sorted(aggr_logs):
        print '%s/%s - %s' % (node, log['name'], mb(log['bytes']))
        total += log['bytes']
    print 'Total: %s' % gb(total)


list_logs_parser.set_defaults(func=do_list_logs)


def output_lines(out_q):
    while True:
        line = out_q.get()
        if not line:
            break
        print line


def consumer(in_q, out_q, swift, args):
    while True:
        item = in_q.get()
        if item is None:
            break
        container, obj = item
        for line in swift.iter_object_lines('.logs', container, obj):
            if args.pattern in line:
                out_q.put(line)


def do_grep_logs(swift, args):
    aggr_logs = get_aggr_logs(swift, args)

    out_q = Queue()
    output_worker = threading.Thread(target=output_lines, args=(out_q,))
    output_worker.start()

    in_q = Queue()
    workers = []
    for i in range(args.workers):
        t = threading.Thread(target=consumer, args=(
            in_q, out_q, swift, args))
        t.start()
        workers.append(t)
    for date, node, log in sorted(aggr_logs):
        in_q.put((node, log['name']))
    for t in workers:
        in_q.put(None)
    for t in workers:
        t.join()
    out_q.put(None)
    output_worker.join()

grep_logs_parser.set_defaults(func=do_grep_logs)


def main():
    args = parser.parse_args()
    swift = InternalClient(ConfigString(ic_conf_body), 'test', 1)
    args.func(swift, args)


if __name__ == "__main__":
    sys.exit(main())