ceph/qa/tasks/deduplication.py

"""
Run ceph-dedup-tool
"""
import contextlib
import logging
import gevent
from teuthology import misc as teuthology
import json
import time
from io import StringIO

log = logging.getLogger(__name__)

@contextlib.contextmanager
def task(ctx, config):
    """
    Run ceph-dedup-tool.
    The config should be as follows::
        ceph-dedup-tool:
          clients: [client list]
          op: <operation name>
          pool: <pool name>
          chunk_pool: <chunk pool name>
          chunk_size: <chunk size>
          chunk_algorithm: <chunk algorithm, fixed|fastcdc>
          fingerprint_algorithm: <fingerprint algorithm, sha1|sha256|sha512>
          chunk_dedup_threashold: <the number of duplicate chunks to trigger chunk dedup>
          max_thread: <the number of threads>
          wakeup_period: <duration>
    For example::
        tasks:
        - exec:
            client.0:
              - sudo ceph osd pool create low_tier 4
        - deduplication:
            clients: [client.0]
            op: 'sample-dedup'
            pool: 'default.rgw.buckets.data'
            chunk_pool: 'low_tier'
            chunk_size: 131072
            chunk_algorithm: 'fastcdc'
            fingerprint_algorithm: 'sha1'
            chunk_dedup_threshold: 5
            max_thread: 2
            wakeup_period: 20
            sampling_ratio: 100
    """
    log.info('Beginning deduplication...')
    assert isinstance(config, dict), \
        "please list clients to run on"

    args = [
        'ceph-dedup-tool']
    if config.get('op', None):
        args.extend(['--op', config.get('op', None)])
    if config.get('chunk_pool', None):
        args.extend(['--chunk-pool', config.get('chunk_pool', None)])
    if config.get('chunk_size', False):
        args.extend(['--chunk-size', str(config.get('chunk_size', 131072))])
    if config.get('chunk_algorithm', False):
        args.extend(['--chunk-algorithm', config.get('chunk_algorithm', None)] )
    if config.get('fingerprint_algorithm', False):
        args.extend(['--fingerprint-algorithm', config.get('fingerprint_algorithm', None)] )
    if config.get('chunk_dedup_threshold', False):
        args.extend(['--chunk-dedup-threshold', str(config.get('chunk_dedup_threshold', 1))])
    if config.get('max_thread', False):
        args.extend(['--max-thread', str(config.get('max_thread', 2))])
    if config.get('sampling_ratio', False):
        args.extend(['--sampling-ratio', str(config.get('sampling_ratio', 100))])
    if config.get('wakeup_period', False):
        args.extend(['--wakeup-period', str(config.get('wakeup_period', 20))])
    if config.get('pool', False):
        args.extend(['--pool', config.get('pool', None)])

    args.extend([
        '--debug',
        '--daemon',
        '--loop'])

    def thread():
        run_remote(args, False, 0)

    def run_remote(args, need_wait, client_num):
        clients = ['client.{id}'.format(id=id_) for id_ in teuthology.all_roles_of_type(ctx.cluster, 'client')]
        log.info('clients are %s' % clients)
        role = 'client.{id}'.format(id=client_num)
        if role not in clients:
            raise Exception('wrong client {c}'.format(c=role))
        assert isinstance(role, str)
        PREFIX = 'client.'
        assert role.startswith(PREFIX)
        testdir = teuthology.get_testdir(ctx)
        cmd_args = [
                'adjust-ulimits',
                'ceph-coverage',
                '{tdir}/archive/coverage'.format(tdir=testdir)]
        cmd_args.extend(args)
        log.info("cmd: %s", cmd_args)
        tries = 0
        while True:
            (remote,) = ctx.cluster.only(role).remotes.keys()
            proc = remote.run(
                args=cmd_args,
                wait=need_wait, check_status=False,
                stdout=StringIO(),
                )
            log.info('exitstatus {r}'.format(r=proc.exitstatus))
            if proc.exitstatus == 0 or need_wait == False:
                log.info('proc stdout ', proc.stdout.getvalue())
                return proc.stdout.getvalue().strip()
            tries += 1
            if tries > 30:
                raise Exception('timed out getting correct exitstatus')
            time.sleep(30)

    def get_chunk_objs(chunk_pool):
        chunk_obj_list = run_remote(('rados ls -p ' + chunk_pool).split(), True, 1).split()
        if chunk_obj_list == False:
            return None
        else:
            return chunk_obj_list

    def get_ref_list(chunk_pool, chunk_obj):
        # get reference list of chunk object
        dump_str = run_remote(
            ('ceph-dedup-tool --op dump-chunk-refs --chunk-pool '
            + chunk_pool + ' --object ' + chunk_obj).split(),
            True, 1
        )
        # fail in case that reference object is not written
        assert len(dump_str) > 0
        log.info('{0} obj has {1} refs'
            .format(chunk_obj, json.loads(dump_str)['count']))

        # check if chunk object's reference object exists in base-tier
        ref_list = json.loads(dump_str)['refs']
        return ref_list

    # To validate whether the sample-dedup operation works well, this function checks if
    #   1. sample-dedup has been started and
    #   2. reference of chunk objects' exists in correct base pool
    def validate():
        log.info('start validating sample-dedup')
        base_pool = config.get('pool', None)
        chunk_pool = config.get('chunk_pool', None)
        max_validation_cnt = 15
        retry_cnt = 0
        # chunk objs for re-validation after chunk-repair
        retry_chunk_objs = list() 

        # check whether sample-dedup has been started
        chunk_obj_list = get_chunk_objs(chunk_pool)
        while (chunk_obj_list == None or len(chunk_obj_list) == 0) and retry_cnt < max_validation_cnt:
            # retry getting # chunk objs after 30 secs of sleep
            time.sleep(30)
            chunk_obj_list = get_chunk_objs(chunk_pool)
            retry_cnt += 1
            log.info('chunk pool empty. retry ', retry_cnt)
        assert retry_cnt < max_validation_cnt

        log.info('sample-dedup started successfully')

        retry_cnt = 0
        max_validation_cnt = 5
        # validate chunk pool for max_validation_cnt times
        while retry_cnt < max_validation_cnt:
            for chunk_obj in chunk_obj_list:
                ref_list = get_ref_list(chunk_pool, chunk_obj)
                for ref in ref_list:
                    ret = run_remote(
                        ('rados -p ' + base_pool + ' stat ' + ref['oid'])
                        .split(), True, 1
                    )
                    # check if ref exists in base pool
                    if ret == False or len(ret) == 0:
                        # if ref not exists in base pool, try repair in order to avoid 
                        # false-positive inconsistent reference
                        ret = run_remote(('ceph osd pool stats ' + base_pool).split(), True, 1)
                        assert len(ret) > 0
                        base_pool_id = ret.split()[3]
                        ret = run_remote(
                            ('ceph-dedup-tool --op chunk-repair --chunk-pool '
                            + chunk_pool + ' --object ' + chunk_obj + ' --target-ref ' 
                            + ref['oid'] + ' --target-ref-pool-id ' + base_pool_id)
                            .split(), True, 1
                        )
                        retry_chunk_objs.append(chunk_obj)
                    log.info('{0} obj exists in {1}'.format(ref['oid'], base_pool))

            # retry validation for repaired objects
            for chunk_obj in retry_chunk_objs:
                ref_list = get_ref_list(chunk_pool, chunk_obj)
                for ref in ref_list:
                    ret = run_remote(
                        ('rados -p ' + base_pool + ' stat ' + ref['oid'])
                        .split(), True, 1
                    )
                    assert len(ret) > 0
                    log.info(
                        '{0} obj exists in {1} after repair'.format(ref['oid'], 
                        base_pool)
                    )
            retry_chunk_objs = list()

            # get chunk objects for the next loop
            chunk_obj_list = get_chunk_objs(chunk_pool)
            retry_cnt += 1
            time.sleep(30)
        return True


    running = gevent.spawn(thread)
    checker = gevent.spawn(validate)

    try:
        yield
    finally:
        log.info('joining ceph-dedup-tool')
        running.get()
        checker.get()