Troll the disks for async_pendings and gather some stats

.gitignore

/*.json

async_pending_stats.py

#!/usr/bin/env python
import sys
import os
import errno
from argparse import ArgumentParser
from collections import defaultdict
import pickle
import logging
import thread
import threading
from Queue import Queue, Empty
import time
import random
import json

from swift.common.storage_policy import POLICIES
from swift.common.ring import Ring
from swift.obj.diskfile import get_async_dir
from swift.common.utils import RateLimitedIterator, split_path

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


parser = ArgumentParser()
parser.add_argument('devices', help='root of devices tree for node',
                    nargs='*', default=['/srv/node'])
parser.add_argument('--policy-index', help='the policy index',
                    type=int, default=0)
parser.add_argument('--limit', help='max number of asyncs to check per disk',
                    default=None, type=int)
parser.add_argument('--updates-per-second', default=250.0, type=float,
                    help='max number of asyncs to check per second')
parser.add_argument('--top-stats', help='display N top account & container',
                    default=10, type=int)
parser.add_argument('--workers', help='number of workers', type=int,
                    default=24, )
parser.add_argument('--verbose', help='log at debug', action='store_true')
parser.add_argument('--swift-dir', help='y u no use /etc/swift',
                    default='/etc/swift')
parser.add_argument('--json', action='store_true', help='dump raw json stats')


class AtomicStats(object):

    def __init__(self):
        self.stats = defaultdict(int)
        self.lock = threading.RLock()

    def incr(self, key, amount=1):
        with self.lock:
            self.stats[key] += amount

    def __iter__(self):
        return iter(self.stats.items())


STATS = AtomicStats()


def handle_update(update_path, container_ring, args):
    with open(update_path) as f:
        update_data = pickle.load(f)
    if args.verbose:
        logging.debug('Found %s\n%s' % (update_path, json.dumps(
            update_data, indent=2)))
    num_success = len(update_data.get('successes', []))
    container_path = update_data.get('container_path')
    if container_path:
        account, container = split_path('/' + container_path, minsegs=2)
    else:
        account, container = \
            update_data['account'], update_data['container']
    _part, nodes = container_ring.get_nodes(account, container)
    bad_devs = [n['device'] for n in nodes
                if n['id'] not in update_data.get('successes', [])]
    if len(bad_devs) == 1:
        logging.debug('Notice %r waiting on update to %s',
                      update_path, ','.join(bad_devs))
    return {
        'op': update_data['op'],
        'account': account,
        'container': container,
        'num_success': num_success,
        'bad_devs': bad_devs,
    }


def consumer(q, args, ring):
    while True:
        update_path = q.get()
        if update_path is None:
            return
        STATS.incr('count')
        update_data = handle_update(update_path, ring, args)
        update_stats(STATS, update_data)


def update_stats(stats, update):
    stats.incr('op_%s' % update['op'])
    stats.incr('acct_%s' % update['account'])
    key = 'cont_%s/%s' % (update['account'], update['container'])
    stats.incr(key)
    key = 'success_%s' % update['num_success']
    stats.incr(key)
    for dev in update['bad_devs']:
        key = 'dev_%s' % dev
        stats.incr(key)


def _display_stats(stats, args):
    accounts = []
    containers = []
    success_counts = []
    ops = []
    devs = []
    logging.info('=' * 50)
    for k, v in stats:
        if k.startswith('acct_'):
            accounts.append((v, k[5:]))
        elif k.startswith('cont_'):
            containers.append((v, k[5:]))
        elif k.startswith('success_'):
            success_counts.append((k, v))
        elif k.startswith('op_'):
            ops.append((k[3:], v))
        elif k.startswith('dev_'):
            devs.append((v, k[4:]))
        else:
            logging.info('%-9s: %s', k, v)
    for k, v in ops:
        logging.info('%-9s: %s' % (k, v))
    success_counts.sort()
    for k, v in success_counts:
        logging.info('%s: %s', k, v)
    logging.info('-' * 50)
    accounts.sort(reverse=True)
    for v, k in accounts[:args.top_stats]:
        logging.info('%s: %s', k, v)
    containers.sort(reverse=True)
    for v, k in containers[:args.top_stats]:
        logging.info('%s: %s', k, v)
    devs.sort(reverse=True)
    for v, k in devs[:args.top_stats]:
        logging.info('%s: %s', k, v)


def display_stats(q, args):
    while True:
        try:
            q.get(block=False)
        except Empty:
            _display_stats(STATS, args)
            time.sleep(1.0)
        else:
            return


def iter_update_paths(device_path, args):
    policy = POLICIES[args.policy_index]
    asyncdir = get_async_dir(policy)

    num_updates = 0
    async_path = os.path.join(device_path, asyncdir)
    try:
        suffixes = os.listdir(async_path)
    except OSError as e:
        if e.errno == errno.ENOENT:
            return
        else:
            raise
    random.shuffle(suffixes)
    for suffix in suffixes:
        try:
            int(suffix, 16)
        except ValueError:
            continue
        suffix_path = os.path.join(async_path, suffix)
        updates = os.listdir(suffix_path)
        random.shuffle(updates)
        for update in updates:
            num_updates += 1
            if args.limit and num_updates >= args.limit:
                return
            update_path = os.path.join(suffix_path, update)
            yield update_path


def feed_queue(q, device_dir, args):
    update_path_iter = iter_update_paths(device_dir, args)
    for update_path in RateLimitedIterator(
            update_path_iter, args.updates_per_second):
        q.put(update_path)


def main():
    args = parser.parse_args()

    if args.verbose:
        level = logging.DEBUG
    else:
        level = logging.INFO
    logging.basicConfig(level=level)

    container_ring = Ring(os.path.join(args.swift_dir, 'container.ring.gz'))

    stats_kill_q = Queue(1)
    stats_worker = threading.Thread(target=display_stats, args=(
        stats_kill_q, args))
    stats_worker.start()

    q = Queue(1000)
    workers = []
    feeders = []
    try:
        for i in range(args.workers):
            t = threading.Thread(target=consumer, args=(
                q, args, container_ring))
            t.start()
            workers.append(t)
        for device_root in args.devices:
            device_dirs = os.listdir(device_root)
            for device_dir in device_dirs:
                device_path = os.path.join(device_root, device_dir)
                u = threading.Thread(target=feed_queue, args=(
                    q, device_path, args))
                u.start()
                feeders.append(u)
        for u in feeders:
            u.join()
    finally:
        logging.info('queue finished')
        for t in workers:
            q.put(None)
        for t in workers:
            t.join()
    logging.info('workers finished')

    stats_kill_q.put(None)
    stats_worker.join()
    if args.json:
        json.dump(STATS.stats, sys.stdout)
    else:
        _display_stats(STATS, args)


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