<commit_before>from cStringIO import StringIO

import argparse
import contextlib
import logging
import os
import sys

from teuthology import misc as teuthology
from teuthology import contextutil
from teuthology.parallel import parallel
from ..orchestra import run

log = logging.getLogger(__name__)

class DaemonState(object):
    def __init__(self, remote, role, id_, *command_args, **command_kwargs):
        self.remote = remote
        self.command_args = command_args
        self.command_kwargs = command_kwargs
        self.role = role
        self.id_ = id_
        self.log = command_kwargs.get('logger', log)
        self.proc = None

    def stop(self):
        """
        Note: this can raise a run.CommandFailedError,
        run.CommandCrashedError, or run.ConnectionLostError.
        """
        if not self.running():
            self.log.error('tried to stop a non-running daemon')
            return
        self.proc.stdin.close()
        self.log.debug('waiting for process to exit')
        run.wait([self.proc])
        self.proc = None
        self.log.info('Stopped')

    def restart(self, *args, **kwargs):
        self.log.info('Restarting')
        if self.proc is not None:
            self.log.debug('stopping old one...')
            self.stop()
        cmd_args = list(self.command_args)
        cmd_args.extend(args)
        cmd_kwargs = self.command_kwargs
        cmd_kwargs.update(kwargs)
        self.proc = self.remote.run(*cmd_args, **cmd_kwargs)
        self.log.info('Started')

    def running(self):
        return self.proc is not None

    def reset(self):
        self.proc = None


class CephState(object):
    def __init__(self):
        self.daemons = {}

    def add_daemon(self, remote, role, id_, *args, **kwargs):
        if role not in self.daemons:
            self.daemons[role] = {}
        if id_ in self.daemons[role]:
            self.daemons[role][id_].stop()
            self.daemons[role][id_] = None
        self.daemons[role][id_] = DaemonState(remote, role, id_, *args, **kwargs)
        self.daemons[role][id_].restart()

    def get_daemon(self, role, id_):
        if role not in self.daemons:
            return None
        return self.daemons[role].get(str(id_), None)

    def iter_daemons_of_role(self, role):
        return self.daemons.get(role, {}).values()

@contextlib.contextmanager
def ceph_log(ctx, config):
    log.info('Creating log directories...')
    archive_dir = '{tdir}/archive'.format(tdir=teuthology.get_testdir(ctx))
    run.wait(
        ctx.cluster.run(
            args=[
                'install', '-d', '-m0755', '--',
                '{adir}/log'.format(adir=archive_dir),
                '{adir}/log/valgrind'.format(adir=archive_dir),
                '{adir}/profiling-logger'.format(adir=archive_dir),
                ],
            wait=False,
            )
        )

    try:
        yield
    finally:

        if ctx.archive is not None:
            log.info('Compressing logs...')
            run.wait(
                ctx.cluster.run(
                    args=[
                        'find',
                        '{adir}/log'.format(adir=archive_dir),
                        '-name',
                        '*.log',
                        '-print0',
                        run.Raw('|'),
                        'xargs',
                        '-0',
                        '--no-run-if-empty',
                        '--',
                        'gzip',
                        '--',
                        ],
                    wait=False,
                    ),
                )

            # log file transfer is done by the generic archive data
            # handling

@contextlib.contextmanager
def ship_utilities(ctx, config):
    assert config is None
    FILES = ['daemon-helper', 'enable-coredump', 'chdir-coredump',
             'valgrind.supp', 'kcon_most']
    testdir = teuthology.get_testdir(ctx)
    for filename in FILES:
        log.info('Shipping %r...', filename)
        src = os.path.join(os.path.dirname(__file__), filename)
        dst = os.path.join(testdir, filename)
        with file(src, 'rb') as f:
            for rem in ctx.cluster.remotes.iterkeys():
                teuthology.write_file(
                    remote=rem,
                    path=dst,
                    data=f,
                    )
                f.seek(0)
                rem.run(
                    args=[
                        'chmod',
                        'a=rx',
                        '--',
                        dst,
                        ],
                    )

    try:
        yield
    finally:
        log.info('Removing shipped files: %s...', ' '.join(FILES))
        filenames = (
            os.path.join(testdir, filename)
            for filename in FILES
            )
        run.wait(
            ctx.cluster.run(
                args=[
                    'rm',
                    '-rf',
                    '--',
                    ] + list(filenames),
                wait=False,
                ),
            )

def _update_deb_package_list_and_install(remote, debs, branch):
    """
    updates the package list so that apt-get can
    download the appropriate packages
    """

    # run gpg just to initialize
    r = remote.run(
            args=[
                'gpg', '-K',
                ],
                )

    # check for ceph release key
    r = remote.run(
            args=[
                'apt-key', 'list', run.Raw('|'), 'grep', 'Ceph',
                ],
            stdout=StringIO(),
        )
    if r.stdout.getvalue().find('Ceph Release Key') == -1:
        # if it doesn't exist, add it
        remote.run(
                args=[
                    'wget', '-q', '-O-',
                    'https://ceph.com/git/?p=ceph.git;a=blob_plain;f=keys/autobuild.asc',
                    run.Raw('|'),
                    'sudo', 'apt-key', 'add', '-',
                    ],
                stdout=StringIO(),
                )

    # get ubuntu release (precise, quantal, etc.)
    r = remote.run(
            args=['lsb_release', '-sc'],
            stdout=StringIO(),
            )

    out = r.stdout.getvalue().strip()
    log.info("release type:" + out)

    remote.run(
            args=[
                'echo', 'deb',
                'http://gitbuilder.ceph.com/ceph-deb-' + out + '-x86_64-basic/ref/' + branch,
                out, 'main', run.Raw('|'),
                'sudo', 'tee', '/etc/apt/sources.list.d/ceph.list'
                ],
            stdout=StringIO(),
            )
    remote.run(
            args=[
                'sudo', 'apt-get', 'update', run.Raw('&&'),
                'sudo', 'apt-get', '-y', '--force-yes',
                'install',
                ] + debs,
            stdout=StringIO(),
            )

def install_debs(ctx, debs, branch):
    """
    installs Debian packages.
    """
    log.info("Installing ceph debian packages: {debs}".format(debs=', '.join(debs)))
    with parallel() as p:
        for remote in ctx.cluster.remotes.iterkeys():
            p.spawn(_update_deb_package_list_and_install, remote, debs, branch)

def _remove_deb(remote, debs):
    args=[
        'sudo', 'apt-get', '-y', '--force-yes', 'purge',
        ]
    args.extend(debs)
    args.extend([
            run.Raw('||'),
            'true'
            ])<commit_msg>ceph: simplify apt-key management<commit_after>from cStringIO import StringIO

import argparse
import contextlib
import logging
import os
import sys

from teuthology import misc as teuthology
from teuthology import contextutil
from teuthology.parallel import parallel
from ..orchestra import run

log = logging.getLogger(__name__)

class DaemonState(object):
    def __init__(self, remote, role, id_, *command_args, **command_kwargs):
        self.remote = remote
        self.command_args = command_args
        self.command_kwargs = command_kwargs
        self.role = role
        self.id_ = id_
        self.log = command_kwargs.get('logger', log)
        self.proc = None

    def stop(self):
        """
        Note: this can raise a run.CommandFailedError,
        run.CommandCrashedError, or run.ConnectionLostError.
        """
        if not self.running():
            self.log.error('tried to stop a non-running daemon')
            return
        self.proc.stdin.close()
        self.log.debug('waiting for process to exit')
        run.wait([self.proc])
        self.proc = None
        self.log.info('Stopped')

    def restart(self, *args, **kwargs):
        self.log.info('Restarting')
        if self.proc is not None:
            self.log.debug('stopping old one...')
            self.stop()
        cmd_args = list(self.command_args)
        cmd_args.extend(args)
        cmd_kwargs = self.command_kwargs
        cmd_kwargs.update(kwargs)
        self.proc = self.remote.run(*cmd_args, **cmd_kwargs)
        self.log.info('Started')

    def running(self):
        return self.proc is not None

    def reset(self):
        self.proc = None


class CephState(object):
    def __init__(self):
        self.daemons = {}

    def add_daemon(self, remote, role, id_, *args, **kwargs):
        if role not in self.daemons:
            self.daemons[role] = {}
        if id_ in self.daemons[role]:
            self.daemons[role][id_].stop()
            self.daemons[role][id_] = None
        self.daemons[role][id_] = DaemonState(remote, role, id_, *args, **kwargs)
        self.daemons[role][id_].restart()

    def get_daemon(self, role, id_):
        if role not in self.daemons:
            return None
        return self.daemons[role].get(str(id_), None)

    def iter_daemons_of_role(self, role):
        return self.daemons.get(role, {}).values()

@contextlib.contextmanager
def ceph_log(ctx, config):
    log.info('Creating log directories...')
    archive_dir = '{tdir}/archive'.format(tdir=teuthology.get_testdir(ctx))
    run.wait(
        ctx.cluster.run(
            args=[
                'install', '-d', '-m0755', '--',
                '{adir}/log'.format(adir=archive_dir),
                '{adir}/log/valgrind'.format(adir=archive_dir),
                '{adir}/profiling-logger'.format(adir=archive_dir),
                ],
            wait=False,
            )
        )

    try:
        yield
    finally:

        if ctx.archive is not None:
            log.info('Compressing logs...')
            run.wait(
                ctx.cluster.run(
                    args=[
                        'find',
                        '{adir}/log'.format(adir=archive_dir),
                        '-name',
                        '*.log',
                        '-print0',
                        run.Raw('|'),
                        'xargs',
                        '-0',
                        '--no-run-if-empty',
                        '--',
                        'gzip',
                        '--',
                        ],
                    wait=False,
                    ),
                )

            # log file transfer is done by the generic archive data
            # handling

@contextlib.contextmanager
def ship_utilities(ctx, config):
    assert config is None
    FILES = ['daemon-helper', 'enable-coredump', 'chdir-coredump',
             'valgrind.supp', 'kcon_most']
    testdir = teuthology.get_testdir(ctx)
    for filename in FILES:
        log.info('Shipping %r...', filename)
        src = os.path.join(os.path.dirname(__file__), filename)
        dst = os.path.join(testdir, filename)
        with file(src, 'rb') as f:
            for rem in ctx.cluster.remotes.iterkeys():
                teuthology.write_file(
                    remote=rem,
                    path=dst,
                    data=f,
                    )
                f.seek(0)
                rem.run(
                    args=[
                        'chmod',
                        'a=rx',
                        '--',
                        dst,
                        ],
                    )

    try:
        yield
    finally:
        log.info('Removing shipped files: %s...', ' '.join(FILES))
        filenames = (
            os.path.join(testdir, filename)
            for filename in FILES
            )
        run.wait(
            ctx.cluster.run(
                args=[
                    'rm',
                    '-rf',
                    '--',
                    ] + list(filenames),
                wait=False,
                ),
            )

def _update_deb_package_list_and_install(remote, debs, branch):
    """
    updates the package list so that apt-get can
    download the appropriate packages
    """

    # check for ceph release key
    r = remote.run(
        args=[
            'sudo', 'apt-key', 'list', run.Raw('|'), 'grep', 'Ceph',
            ],
        stdout=StringIO(),
        )
    if r.stdout.getvalue().find('Ceph automated package') == -1:
        # if it doesn't exist, add it
        remote.run(
                args=[
                    'wget', '-q', '-O-',
                    'https://ceph.com/git/?p=ceph.git;a=blob_plain;f=keys/autobuild.asc',
                    run.Raw('|'),
                    'sudo', 'apt-key', 'add', '-',
                    ],
                stdout=StringIO(),
                )

    # get ubuntu release (precise, quantal, etc.)
    r = remote.run(
            args=['lsb_release', '-sc'],
            stdout=StringIO(),
            )

    out = r.stdout.getvalue().strip()
    log.info("release type:" + out)

    remote.run(
        args=[
            'echo', 'deb',
            'http://gitbuilder.ceph.com/ceph-deb-' + out + '-x86_64-basic/ref/' + branch,
            out, 'main', run.Raw('|'),
            'sudo', 'tee', '/etc/apt/sources.list.d/ceph.list'
            ],
        stdout=StringIO(),
        )
    remote.run(
        args=[
            'sudo', 'apt-get', 'update', run.Raw('&&'),
            'sudo', 'apt-get', '-y', '--force-yes',
            'install',
            ] + debs,
        stdout=StringIO(),
        )

def install_debs(ctx, debs, branch):
    """
    installs Debian packages.
    """
    log.info("Installing ceph debian packages: {debs}".format(debs=', '.join(debs)))
    with parallel() as p:
        for remote in ctx.cluster.remotes.iterkeys():
            p.spawn(_update_deb_package_list_and_install, remote, debs, branch)

def _remove_deb(remote, debs):
    args=[
        'sudo', 'apt-get', '-y', '--force-yes', 'purge',
        ]
    args.extend(debs)
    args.extend([
            run.Raw('||'),
            'true'
            ])
{'fraction_non_alphanumeric': 0.13304544519843717, 'fraction_numerical': 0.001645075056549455, 'mean_word_length': 1.411171707155842, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1425578', 'n_tokens_mistral': 4398, 'n_tokens_neox': 4196, 'n_words': 882}