funtests/stress moved to new package celery/cyanide

funtests/stress/README.rst

@@ -2,176 +2,7 @@
  Celery Stresstest Suite
 =========================
 
-.. contents::
-    :local:
+The stress test suite project has been moved to a dedicated
+repository here:
 
-Introduction
-============
-
-These tests will attempt to break the worker in different ways.
-
-The worker must currently be started separately, and it's encouraged
-to run the stresstest with different configuration values.
-
-Ideas include:
-
-1)  Frequent maxtasksperchild, single process
-
-::
-
-    $ celery -A stress worker -c 1 --maxtasksperchild=1
-
-2) Frequent scale down & maxtasksperchild, single process
-
-::
-
-    $ AUTOSCALE_KEEPALIVE=0.01 celery -A stress worker --autoscale=1,0 \
-                                                       --maxtasksperchild=1
-
-3) Frequent maxtasksperchild, multiple processes
-
-::
-
-    $ celery -A stress worker -c 8 --maxtasksperchild=1``
-
-4) Default, single process
-
-::
-
-    $ celery -A stress worker -c 1
-
-5) Default, multiple processes
-
-::
-
-    $ celery -A stress worker -c 8
-
-6) Processes termianted by time limits
-
-::
-
-    $ celery -A stress worker --time-limit=1
-
-7) Frequent maxtasksperchild, single process with late ack.
-
-::
-
-    $ celery -A stress worker -c1 --maxtasksperchild=1 -Z acks_late
-
-
-8) Worker using eventlet pool.
-
-    Start the worker::
-
-        $ celery -A stress worker -c1000 -P eventlet
-
-    Then must use the `-g green` test group::
-
-        $ python -m stress -g green
-
-9) Worker using gevent pool.
-
-It's also a good idea to include the ``--purge`` argument to clear out tasks from
-previous runs.
-
-Note that the stress client will probably hang if the test fails, so this
-test suite is currently not suited for automatic runs.
-
-Configuration Templates
------------------------
-
-You can select a configuration template using the `-Z` command-line argument
-to any :program:`celery -A stress` command or the :program:`python -m stress`
-command when running the test suite itself.
-
-The templates available are:
-
-* default
-
-    Using amqp as a broker and rpc as a result backend,
-    and also using json for task and result messages.
-
-* redis
-
-    Using redis as a broker and result backend
-
-* acks_late
-
-    Enables late ack globally.
-
-* pickle
-
-    Using pickle as the serializer for tasks and results
-    (also allowing the worker to receive and process pickled messages)
-
-
-You can see the resulting configuration from any template by running
-the command::
-
-    $ celery -A stress report -Z redis
-
-
-Example running the stress test using the ``redis`` configuration template::
-
-    $ python -m stress -Z redis
-
-Example running the worker using the ``redis`` configuration template::
-
-    $ celery -A stress worker -Z redis
-
-
-You can also mix several templates by listing them separated by commas::
-
-    $ celery -A stress worker -Z redis,acks_late
-
-In this example (``redis,acks_late``) the ``redis`` template will be used
-as a configuration, and then additional keys from the ``acks_late`` template
-will be added on top as changes::
-
-    $ celery -A stress report -Z redis,acks_late,pickle
-
-Running the client
-------------------
-
-After the worker is running you can start the client to run the complete test
-suite::
-
-    $ python -m stress
-
-You can also specify which tests to run:
-
-    $ python -m stress revoketermfast revoketermslow
-
-Or you can start from an offset, e.g. to skip the two first tests use
-``--offset=2``::
-
-    $ python -m stress --offset=2
-
-See ``python -m stress --help`` for a list of all available options.
-
-
-Options
-=======
-
-Using a different broker
-------------------------
-You can set the environment ``CSTRESS_BROKER`` to change the broker used::
-
-    $ CSTRESS_BROKER='amqp://' celery -A stress worker # …
-    $ CSTRESS_BROKER='amqp://' python -m stress
-
-Using a different result backend
---------------------------------
-
-You can set the environment variable ``CSTRESS_BACKEND`` to change
-the result backend used::
-
-    $ CSTRESS_BACKEND='amqp://' celery -A stress worker # …
-    $ CSTRESS_BACKEND='amqp://' python -m stress
-
-Using a custom queue
---------------------
-
-A queue named ``c.stress`` is created and used by default,
-but you can change the name of this queue using the ``CSTRESS_QUEUE``
-environment variable.
+https://github.com/celery/cyanide/

funtests/stress/rabbit-restart-loop.sh

@@ -1,13 +0,0 @@
-#!/usr/bin/env bash
-
-secs=${1:-30}
-secs=$((secs - 1))
-
-while true; do
-    sudo rabbitmqctl start_app
-    echo "sleep for ${secs}s"
-    sleep $secs
-    sudo rabbitmqctl stop_app
-    echo "sleep for 1s"
-    sleep 1
-done

funtests/stress/run/Vagrantfile

@@ -1,125 +0,0 @@
-# -*- mode: ruby -*-
-# vi: set ft=ruby :
-
-# Vagrantfile API/syntax version. Don't touch unless you know what you're doing!
-VAGRANTFILE_API_VERSION = "2"
-
-Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
-  # All Vagrant configuration is done here. The most common configuration
-  # options are documented and commented below. For a complete reference,
-  # please see the online documentation at vagrantup.com.
-
-  # Every Vagrant virtual environment requires a box to build off of.
-  config.vm.box = "ubuntu/trusty64"
-
-  config.vm.provision :shell, path: "provision/provision.sh",
-    privileged: true
-
-  # Disable automatic box update checking. If you disable this, then
-  # boxes will only be checked for updates when the user runs
-  # `vagrant box outdated`. This is not recommended.
-  # config.vm.box_check_update = false
-
-  # Create a forwarded port mapping which allows access to a specific port
-  # within the machine from a port on the host machine. In the example below,
-  # accessing "localhost:8080" will access port 80 on the guest machine.
-  # config.vm.network "forwarded_port", guest: 80, host: 8080
-
-  # Create a private network, which allows host-only access to the machine
-  # using a specific IP.
-  config.vm.network "private_network", ip: "192.168.33.123"
-
-  # Create a public network, which generally matched to bridged network.
-  # Bridged networks make the machine appear as another physical device on
-  # your network.
-  # config.vm.network "public_network"
-
-  # If true, then any SSH connections made will enable agent forwarding.
-  # Default value: false
-  # config.ssh.forward_agent = true
-
-  # Share an additional folder to the guest VM. The first argument is
-  # the path on the host to the actual folder. The second argument is
-  # the path on the guest to mount the folder. And the optional third
-  # argument is a set of non-required options.
-  # config.vm.synced_folder "../data", "/vagrant_data"
-
-  # Provider-specific configuration so you can fine-tune various
-  # backing providers for Vagrant. These expose provider-specific options.
-  # Example for VirtualBox:
-  #
-  config.vm.provider "virtualbox" do |vb|
-  #   # Don't boot with headless mode
-  #   vb.gui = true
-  #
-  #   # Use VBoxManage to customize the VM. For example to change memory:
-    vb.customize ["modifyvm", :id, "--memory", "1024"]
-  end
-  #
-  # View the documentation for the provider you're using for more
-  # information on available options.
-
-  # Enable provisioning with CFEngine. CFEngine Community packages are
-  # automatically installed. For example, configure the host as a
-  # policy server and optionally a policy file to run:
-  #
-  # config.vm.provision "cfengine" do |cf|
-  #   cf.am_policy_hub = true
-  #   # cf.run_file = "motd.cf"
-  # end
-  #
-  # You can also configure and bootstrap a client to an existing
-  # policy server:
-  #
-  # config.vm.provision "cfengine" do |cf|
-  #   cf.policy_server_address = "10.0.2.15"
-  # end
-
-  # Enable provisioning with Puppet stand alone.  Puppet manifests
-  # are contained in a directory path relative to this Vagrantfile.
-  # You will need to create the manifests directory and a manifest in
-  # the file default.pp in the manifests_path directory.
-  #
-  # config.vm.provision "puppet" do |puppet|
-  #   puppet.manifests_path = "manifests"
-  #   puppet.manifest_file  = "site.pp"
-  # end
-
-  # Enable provisioning with chef solo, specifying a cookbooks path, roles
-  # path, and data_bags path (all relative to this Vagrantfile), and adding
-  # some recipes and/or roles.
-  #
-  # config.vm.provision "chef_solo" do |chef|
-  #   chef.cookbooks_path = "../my-recipes/cookbooks"
-  #   chef.roles_path = "../my-recipes/roles"
-  #   chef.data_bags_path = "../my-recipes/data_bags"
-  #   chef.add_recipe "mysql"
-  #   chef.add_role "web"
-  #
-  #   # You may also specify custom JSON attributes:
-  #   chef.json = { :mysql_password => "foo" }
-  # end
-
-  # Enable provisioning with chef server, specifying the chef server URL,
-  # and the path to the validation key (relative to this Vagrantfile).
-  #
-  # The Opscode Platform uses HTTPS. Substitute your organization for
-  # ORGNAME in the URL and validation key.
-  #
-  # If you have your own Chef Server, use the appropriate URL, which may be
-  # HTTP instead of HTTPS depending on your configuration. Also change the
-  # validation key to validation.pem.
-  #
-  # config.vm.provision "chef_client" do |chef|
-  #   chef.chef_server_url = "https://api.opscode.com/organizations/ORGNAME"
-  #   chef.validation_key_path = "ORGNAME-validator.pem"
-  # end
-  #
-  # If you're using the Opscode platform, your validator client is
-  # ORGNAME-validator, replacing ORGNAME with your organization name.
-  #
-  # If you have your own Chef Server, the default validation client name is
-  # chef-validator, unless you changed the configuration.
-  #
-  #   chef.validation_client_name = "ORGNAME-validator"
-end

funtests/stress/run/provision/celeryd-init.config

@@ -1,12 +0,0 @@
-CELERYD_NODES="worker1"
-CELERY_BIN="/usr/local/bin/celery"
-CELERY_APP="stress"
-CELERYD_CHDIR="/opt/devel/stress"
-CELERYD_OPTS="-c10 --maxtasksperchild=256 -Z vagrant1"
-CELERYD_LOG_FILE="/var/log/celery/%n%I.log"
-CELERYD_PID_FILE="/var/run/celery/%n.pid"
-
-CELERYD_USER="celery"
-CELERYD_GROUP="celery"
-
-CELERY_CREATE_DIRS=1

funtests/stress/run/provision/provision.sh

@@ -1,199 +0,0 @@
-#!/bin/bash
-
-APT_SOURCES_LST="/etc/apt/sources.list.d/"
-
-DEVEL_DIR="/opt/devel"
-
-WGET="wget"
-RABBITMQCTL="rabbitmqctl"
-
-RABBITMQ_APT_URL="http://www.rabbitmq.com/debian/"
-RABBITMQ_APT_VER="testing main"
-RABBITMQ_APT_KEY="https://www.rabbitmq.com/rabbitmq-signing-key-public.asc"
-RABBITMQ_DEB="rabbitmq-server"
-
-RABBITMQ_USERNAME="testing"
-RABBITMQ_PASSWORD="t3s71ng"
-RABBITMQ_VHOST="/testing"
-
-REDIS_DEB="redis-server"
-REDIS_CONF="/etc/redis/redis.conf"
-
-GIT_ROOT="${DEVEL_DIR}"
-
-GITHUB_ROOT="https://github.com/"
-CELERY_GITHUB_USER="celery"
-CELERY_USER="celery"
-CELERY_GROUP="celery"
-CELERY_DIR="${GIT_ROOT}/celery"
-CELERY_FUNTESTS="${CELERY_DIR}/funtests/stress"
-CELERY_CONFIG_SRC="${CELERY_FUNTESTS}/run/provision/celeryd-init.config"
-CELERY_CONFIG_DST="/etc/default/celeryd"
-STRESS_DIR="${GIT_ROOT}/stress"
-
-
-die () {
-    echo $*
-    exit 1
-}
-
-# --- grent
-
-add_real_user () {
-    user_shell=${3:-/bin/bash}
-    addgroup $2
-    echo creating user "$1 group='$2' shell='${user_shell}'"
-    echo | adduser -q "$1" --shell="${user_shell}"   \
-            --ingroup="$2"                           \
-            --disabled-password  1>/dev/null 2>&1
-    id "$1" || die "Not able to create user"
-}
-
-# --- system
-
-make_directories () {
-    mkdir -p "${DEVEL_DIR}"
-}
-
-enable_bash_vi_mode () {
-    echo "set -o vi" >> /etc/bash.bashrc
-}
-
-configure_system () {
-    make_directories
-    enable_bash_vi_mode
-}
-
-
-# --- apt
-
-apt_update() {
-    apt-get update
-}
-
-add_apt_source () {
-    echo "deb $1" >> "${APT_SOURCES_LST}/rabbitmq.list"
-}
-
-add_apt_key() {
-    "$WGET" --quiet -O - "$1" | apt-key add -
-}
-
-apt_install () {
-    apt-get install -y "$1"
-}
-
-# --- rabbitmq
-
-rabbitmq_add_user () {
-    "$RABBITMQCTL" add_user "$1" "$2"
-}
-
-rabbitmq_add_vhost () {
-    "$RABBITMQCTL" add_vhost "$1"
-}
-
-rabbitmq_set_perm () {
-    "$RABBITMQCTL" set_permissions -p $1 $2 '.*' '.*' '.*'
-}
-
-install_rabbitmq() {
-    add_apt_source "${RABBITMQ_APT_URL} ${RABBITMQ_APT_VER}"
-    add_apt_key "${RABBITMQ_APT_KEY}"
-    apt_update
-    apt_install "${RABBITMQ_DEB}"
-
-    rabbitmq_add_user "${RABBITMQ_USERNAME}" "${RABBITMQ_PASSWORD}"
-    rabbitmq_add_vhost "${RABBITMQ_VHOST}"
-    rabbitmq_set_perm "${RABBITMQ_VHOST}" "${RABBITMQ_USERNAME}"
-}
-
-# --- redis
-
-restart_redis () {
-    service redis-server restart
-}
-
-
-install_redis () {
-    apt_install "${REDIS_DEB}"
-    sed -i 's/^bind .*$/#bind 127.0.0.1/' "${REDIS_CONF}"
-    restart_redis
-}
-
-# --- git
-
-install_git () {
-    apt_install git
-}
-
-
-github_clone () {
-    mkdir "${CELERY_DIR}"
-    chown "${CELERY_USER}" "${CELERY_DIR}"
-    (cd "${GIT_ROOT}"; sudo -u celery git clone "${GITHUB_ROOT}/${1}/${2}")
-}
-
-# --- pip
-
-pip_install () {
-    pip install -U "$1"
-}
-
-install_pip () {
-    apt_install python-setuptools
-    easy_install pip
-    pip_install virtualenv
-    apt_install python-dev
-    pip_install setproctitle
-}
-
-# --- celery
-
-restart_celery () {
-    service celeryd restart
-}
-
-
-install_celery_service () {
-    cp "${CELERY_DIR}/extra/generic-init.d/celeryd" /etc/init.d/
-    chmod +x "/etc/init.d/celeryd"
-    update-rc.d celeryd defaults
-    echo "cp \'${CELERY_CONFIG_SRC}\' \'${CELERY_CONFIG_DST}'"
-    cp "${CELERY_CONFIG_SRC}" "${CELERY_CONFIG_DST}"
-    update-rc.d celeryd enable
-    restart_celery
-}
-
-install_celery () {
-    pip_install celery
-    add_real_user "${CELERY_USER}" "${CELERY_GROUP}"
-    echo github_clone "'${CELERY_GITHUB_USER}'" "'celery'"
-    github_clone "${CELERY_GITHUB_USER}" celery
-    (cd ${CELERY_DIR}; pip install -r requirements/dev.txt);
-    (cd ${CELERY_DIR}; python setup.py develop);
-}
-
-install_stress () {
-    mkdir "${STRESS_DIR}"
-    chown "${CELERY_USER}" "${STRESS_DIR}"
-    cp -r ${CELERY_DIR}/funtests/stress/* "${STRESS_DIR}/"
-}
-
-# --- MAIN
-
-provision () {
-    apt_update
-    configure_system
-    apt_install powertop
-    apt_install htop
-    install_git
-    install_rabbitmq
-    install_redis
-    install_pip
-    install_celery
-    install_stress
-    install_celery_service
-}
-
-provision

funtests/stress/stress/__init__.py

@@ -1,20 +0,0 @@
-# -*- coding: utf-8 -*-
-from __future__ import absolute_import, unicode_literals
-
-import os
-import time
-
-from .data import install_json  # noqa
-
-if os.environ.get('C_SLEEP'):
-
-    _orig_sleep = time.sleep
-
-    def _sleep(n):
-        print('warning: time sleep for {0}s'.format(n))
-        import traceback
-        traceback.print_stack()
-        _orig_sleep(n)
-    time.sleep = _sleep
-
-from .app import app  # noqa

funtests/stress/stress/__main__.py

@@ -1,43 +0,0 @@
-from __future__ import absolute_import, print_function, unicode_literals
-
-from celery.bin.base import Command, Option
-
-from .app import app
-from .suite import Suite
-
-
-class Stress(Command):
-
-    def run(self, *names, **options):
-        try:
-            return Suite(
-                self.app,
-                block_timeout=options.get('block_timeout'),
-            ).run(names, **options)
-        except KeyboardInterrupt:
-            pass
-
-    def get_options(self):
-        return (
-            Option('-i', '--iterations', type='int', default=50,
-                   help='Number of iterations for each test'),
-            Option('-n', '--numtests', type='int', default=None,
-                   help='Number of tests to execute'),
-            Option('-o', '--offset', type='int', default=0,
-                   help='Start at custom offset'),
-            Option('--block-timeout', type='int', default=30 * 60),
-            Option('-l', '--list', action='store_true', dest='list_all',
-                   help='List all tests'),
-            Option('-r', '--repeat', type='float', default=0,
-                   help='Number of times to repeat the test suite'),
-            Option('-g', '--group', default='all',
-                   help='Specify test group (all|green|redis)'),
-            Option('--diag', default=False, action='store_true',
-                   help='Enable diagnostics (slow)'),
-            Option('-J', '--no-join', default=False, action='store_true',
-                   help='Do not wait for task results'),
-        )
-
-
-if __name__ == '__main__':
-    Stress(app=app).execute_from_commandline()

funtests/stress/stress/app.py

@@ -1,177 +0,0 @@
-# -*- coding: utf-8 -*-
-from __future__ import absolute_import, print_function, unicode_literals
-
-import celery
-import os
-import sys
-import signal
-
-from time import sleep
-
-from celery import signals
-from celery.bin.base import Option
-from celery.exceptions import SoftTimeLimitExceeded
-from celery.utils.log import get_task_logger
-
-from .templates import use_template, template_names
-
-logger = get_task_logger(__name__)
-
-IS_CELERY_4 = celery.VERSION[0] >= 4
-
-
-class App(celery.Celery):
-    template_selected = False
-
-    def __init__(self, *args, **kwargs):
-        self.template = kwargs.pop('template', None)
-        super(App, self).__init__(*args, **kwargs)
-        self.user_options['preload'].add(
-            Option(
-                '-Z', '--template', default='default',
-                help='Configuration template to use: {0}'.format(
-                    template_names(),
-                ),
-            )
-        )
-        signals.user_preload_options.connect(self.on_preload_parsed)
-        if IS_CELERY_4:
-            self.on_configure.connect(self._maybe_use_default_template)
-
-    def on_preload_parsed(self, options=None, **kwargs):
-        self.use_template(options['template'])
-
-    def use_template(self, name='default'):
-        if self.template_selected:
-            raise RuntimeError('App already configured')
-        use_template(self, name)
-        self.template_selected = True
-
-    def _maybe_use_default_template(self, **kwargs):
-        if not self.template_selected:
-            self.use_template('default')
-
-    if not IS_CELERY_4:
-        after_configure = None
-
-        def _get_config(self):
-            ret = super(App, self)._get_config()
-            if self.after_configure:
-                self.after_configure(ret)
-            return ret
-
-        def on_configure(self):
-            self._maybe_use_default_template()
-
-app = App('stress', set_as_current=False)
-
-
-@app.task
-def _marker(s, sep='-'):
-    print('{0} {1} {2}'.format(sep * 3, s, sep * 3))
-
-
-@app.task
-def add(x, y):
-    return x + y
-
-
-@app.task(bind=True)
-def ids(self, i):
-    return (self.request.root_id, self.request.parent_id, i)
-
-
-@app.task(bind=True)
-def collect_ids(self, ids, i):
-    return ids, (self.request.root_id, self.request.parent_id, i)
-
-
-@app.task
-def xsum(x):
-    return sum(x)
-
-
-@app.task
-def any_(*args, **kwargs):
-    wait = kwargs.get('sleep')
-    if wait:
-        sleep(wait)
-
-
-@app.task
-def any_returning(*args, **kwargs):
-    any_(*args, **kwargs)
-    return args, kwargs
-
-
-@app.task
-def exiting(status=0):
-    sys.exit(status)
-
-
-@app.task
-def kill(sig=getattr(signal, 'SIGKILL', None) or signal.SIGTERM):
-    os.kill(os.getpid(), sig)
-
-
-@app.task
-def sleeping(i, **_):
-    sleep(i)
-
-
-@app.task
-def sleeping_ignore_limits(i):
-    try:
-        sleep(i)
-    except SoftTimeLimitExceeded:
-        sleep(i)
-
-
-@app.task(bind=True)
-def retries(self):
-    if not self.request.retries:
-        raise self.retry(countdown=1)
-    return 10
-
-
-@app.task
-def print_unicode():
-    logger.warning('hå它 valmuefrø')
-    print('hiöäüß')
-
-
-@app.task
-def segfault():
-    import ctypes
-    ctypes.memset(0, 0, 1)
-    assert False, 'should not get here'
-
-
-@app.task(bind=True)
-def chord_adds(self, x):
-    self.add_to_chord(add.s(x, x))
-    return 42
-
-
-@app.task(bind=True)
-def chord_replace(self, x):
-    return self.replace_in_chord(add.s(x, x))
-
-
-@app.task
-def raising(exc=KeyError()):
-    raise exc
-
-
-@app.task
-def logs(msg, p=False):
-    print(msg) if p else logger.info(msg)
-
-
-def marker(s, sep='-'):
-    print('{0}{1}'.format(sep, s))
-    while True:
-        try:
-            return _marker.delay(s, sep)
-        except Exception as exc:
-            print('Retrying marker.delay(). It failed to start: %s' % exc)

funtests/stress/stress/data.py

@@ -1,81 +0,0 @@
-# -*- coding: utf-8 -*-
-from __future__ import absolute_import, unicode_literals
-
-from celery.five import python_2_unicode_compatible
-
-try:
-    import simplejson as json
-except ImportError:
-    import json  # noqa
-
-type_registry = {}
-
-
-class JSONEncoder(json.JSONEncoder):
-
-    def default(self, obj):
-        try:
-            return super(JSONEncoder, self).default(obj)
-        except TypeError:
-            reducer = getattr(obj, '__to_json__', None)
-            if reducer:
-                return reducer()
-            raise
-
-
-def decode_hook(d):
-    try:
-        d = d['py/obj']
-    except KeyError:
-        return d
-    type_registry[d['type']](**d['attrs'])
-
-
-def install_json():
-    json._default_encoder = JSONEncoder()
-    json._default_decoder.object_hook = decode_hook
-install_json()  # ugh, ugly but it's a test suite after all
-
-
-# this imports kombu.utils.json, so can only import after install_json()
-from celery.utils.debug import humanbytes  # noqa
-from celery.utils.imports import qualname  # noqa
-
-
-def json_reduce(obj, attrs):
-    return {'py/obj': {'type': qualname(obj), 'attrs': attrs}}
-
-
-def jsonable(cls):
-    type_registry[qualname(cls)] = cls.__from_json__
-    return cls
-
-
-@jsonable
-@python_2_unicode_compatible
-class Data(object):
-
-    def __init__(self, label, data):
-        self.label = label
-        self.data = data
-
-    def __str__(self):
-        return '<Data: {0} ({1})>'.format(
-            self.label, humanbytes(len(self.data)),
-        )
-
-    def __repr__(self):
-        return str(self)
-
-    def __to_json__(self):
-        return json_reduce(self, {'label': self.label, 'data': self.data})
-
-    @classmethod
-    def __from_json__(cls, label=None, data=None, **kwargs):
-        return cls(label, data)
-
-    def __reduce__(self):
-        return Data, (self.label, self.data)
-
-BIG = Data('BIG', 'x' * 2 ** 20 * 8)
-SMALL = Data('SMALL', 'e' * 1024)

funtests/stress/stress/fbi.py

@@ -1,66 +0,0 @@
-from __future__ import absolute_import, print_function, unicode_literals
-
-import socket
-import sys
-
-from contextlib import contextmanager
-
-from celery import states
-
-
-class FBI(object):
-
-    def __init__(self, app):
-        self.app = app
-        self.receiver = None
-        self.state = self.app.events.State()
-        self.connection = None
-        self.enabled = False
-
-    def enable(self, enabled):
-        self.enabled = enabled
-
-    @contextmanager
-    def investigation(self):
-        if self.enabled:
-            with self.app.connection() as conn:
-                receiver = self.app.events.Receiver(
-                    conn, handlers={'*': self.state.event},
-                )
-                with receiver.consumer_context() as (conn, _, _):
-                    self.connection = conn
-                    try:
-                        yield self
-                    finally:
-                        self.ffwd()
-        else:
-            yield
-
-    def ffwd(self):
-        while 1:
-            try:
-                self.connection.drain_events(timeout=1)
-            except socket.error:
-                break
-
-    def state_of(self, tid):
-        try:
-            task = self.state.tasks[tid]
-        except KeyError:
-            return 'No events for {0}'.format(tid)
-
-        if task.state in states.READY_STATES:
-            return 'Task {0.uuid} completed with {0.state}'.format(task)
-        elif task.state in states.UNREADY_STATES:
-            return 'Task {0.uuid} waiting in {0.state} state'.format(task)
-        else:
-            return 'Task {0.uuid} in other state {0.state}'.format(task)
-
-    def query(self, ids):
-        return self.app.control.inspect().query_task(id)
-
-    def diag(self, ids, file=sys.stderr):
-        if self.enabled:
-            self.ffwd()
-            for tid in ids:
-                print(self.state_of(tid), file=file)

funtests/stress/stress/suite.py

@@ -1,460 +0,0 @@
-from __future__ import absolute_import, print_function, unicode_literals
-
-import inspect
-import platform
-import random
-import socket
-import sys
-
-from collections import OrderedDict, defaultdict, namedtuple
-from itertools import count
-from time import sleep
-
-from celery import VERSION_BANNER, chain, group, uuid
-from celery.exceptions import TimeoutError
-from celery.five import items, monotonic, range, values
-from celery.utils.debug import blockdetection
-from celery.utils.text import pluralize, truncate
-from celery.utils.timeutils import humanize_seconds
-
-from .app import (
-    marker, _marker, add, any_, collect_ids, exiting, ids, kill, sleeping,
-    sleeping_ignore_limits, any_returning, print_unicode,
-)
-from .data import BIG, SMALL
-from .fbi import FBI
-
-
-BANNER = """\
-Celery stress-suite v{version}
-
-{platform}
-
-[config]
-.> broker: {conninfo}
-
-[toc: {total} {TESTS} total]
-{toc}
-"""
-
-F_PROGRESS = """\
-{0.index}: {0.test.__name__}({0.iteration}/{0.total_iterations}) \
-rep#{0.repeats} runtime: {runtime}/{elapsed} \
-"""
-
-Progress = namedtuple('Progress', (
-    'test', 'iteration', 'total_iterations',
-    'index', 'repeats', 'runtime', 'elapsed', 'completed',
-))
-
-
-Inf = float('Inf')
-
-
-def assert_equal(a, b):
-    assert a == b, '{0!r} != {1!r}'.format(a, b)
-
-
-class StopSuite(Exception):
-    pass
-
-
-def pstatus(p):
-    runtime = monotonic() - p.runtime
-    elapsed = monotonic() - p.elapsed
-    return F_PROGRESS.format(
-        p,
-        runtime=humanize_seconds(runtime, now=runtime),
-        elapsed=humanize_seconds(elapsed, now=elapsed),
-    )
-
-
-class Speaker(object):
-
-    def __init__(self, gap=5.0):
-        self.gap = gap
-        self.last_noise = monotonic() - self.gap * 2
-
-    def beep(self):
-        now = monotonic()
-        if now - self.last_noise >= self.gap:
-            self.emit()
-            self.last_noise = now
-
-    def emit(self):
-        print('\a', file=sys.stderr, end='')
-
-
-def testgroup(*funs):
-    return OrderedDict((fun.__name__, fun) for fun in funs)
-
-
-class BaseSuite(object):
-
-    def __init__(self, app, block_timeout=30 * 60):
-        self.app = app
-        self.connerrors = self.app.connection().recoverable_connection_errors
-        self.block_timeout = block_timeout
-        self.progress = None
-        self.speaker = Speaker()
-        self.fbi = FBI(app)
-        self.init_groups()
-
-    def init_groups(self):
-        acc = defaultdict(list)
-        for attr in dir(self):
-            if not _is_descriptor(self, attr):
-                meth = getattr(self, attr)
-                try:
-                    groups = meth.__func__.__testgroup__
-                except AttributeError:
-                    pass
-                else:
-                    for g in groups:
-                        acc[g].append(meth)
-        # sort the tests by the order in which they are defined in the class
-        for g in values(acc):
-            g[:] = sorted(g, key=lambda m: m.__func__.__testsort__)
-        self.groups = dict(
-            (name, testgroup(*tests)) for name, tests in items(acc)
-        )
-
-    def run(self, names=None, iterations=50, offset=0,
-            numtests=None, list_all=False, repeat=0, group='all',
-            diag=False, no_join=False, **kw):
-        self.no_join = no_join
-        self.fbi.enable(diag)
-        tests = self.filtertests(group, names)[offset:numtests or None]
-        if list_all:
-            return print(self.testlist(tests))
-        print(self.banner(tests))
-        print('+ Enabling events')
-        self.app.control.enable_events()
-        it = count() if repeat == Inf else range(int(repeat) or 1)
-        for i in it:
-            marker(
-                'Stresstest suite start (repetition {0})'.format(i + 1),
-                '+',
-            )
-            for j, test in enumerate(tests):
-                self.runtest(test, iterations, j + 1, i + 1)
-            marker(
-                'Stresstest suite end (repetition {0})'.format(i + 1),
-                '+',
-            )
-
-    def filtertests(self, group, names):
-        tests = self.groups[group]
-        try:
-            return ([tests[n] for n in names] if names
-                    else list(values(tests)))
-        except KeyError as exc:
-            raise KeyError('Unknown test name: {0}'.format(exc))
-
-    def testlist(self, tests):
-        return ',\n'.join(
-            '.> {0}) {1}'.format(i + 1, t.__name__)
-            for i, t in enumerate(tests)
-        )
-
-    def banner(self, tests):
-        app = self.app
-        return BANNER.format(
-            app='{0}:0x{1:x}'.format(app.main or '__main__', id(app)),
-            version=VERSION_BANNER,
-            conninfo=app.connection().as_uri(),
-            platform=platform.platform(),
-            toc=self.testlist(tests),
-            TESTS=pluralize(len(tests), 'test'),
-            total=len(tests),
-        )
-
-    def runtest(self, fun, n=50, index=0, repeats=1):
-        n = getattr(fun, '__iterations__', None) or n
-        print('{0}: [[[{1}({2})]]]'.format(repeats, fun.__name__, n))
-        with blockdetection(self.block_timeout):
-            with self.fbi.investigation():
-                runtime = elapsed = monotonic()
-                i = 0
-                failed = False
-                self.progress = Progress(
-                    fun, i, n, index, repeats, elapsed, runtime, 0,
-                )
-                _marker.delay(pstatus(self.progress))
-                try:
-                    for i in range(n):
-                        runtime = monotonic()
-                        self.progress = Progress(
-                            fun, i + 1, n, index, repeats, runtime, elapsed, 0,
-                        )
-                        try:
-                            fun()
-                        except StopSuite:
-                            raise
-                        except Exception as exc:
-                            print('-> {0!r}'.format(exc))
-                            import traceback
-                            print(traceback.format_exc())
-                            print(pstatus(self.progress))
-                        else:
-                            print(pstatus(self.progress))
-                except Exception:
-                    failed = True
-                    self.speaker.beep()
-                    raise
-                finally:
-                    print('{0} {1} iterations in {2}'.format(
-                        'failed after' if failed else 'completed',
-                        i + 1, humanize_seconds(monotonic() - elapsed),
-                    ))
-                    if not failed:
-                        self.progress = Progress(
-                            fun, i + 1, n, index, repeats, runtime, elapsed, 1,
-                        )
-
-    def missing_results(self, r):
-        return [res.id for res in r if res.id not in res.backend._cache]
-
-    def join(self, r, propagate=False, max_retries=10, **kwargs):
-        if self.no_join:
-            return
-        received = []
-
-        def on_result(task_id, value):
-            received.append(task_id)
-
-        for i in range(max_retries) if max_retries else count(0):
-            received[:] = []
-            try:
-                return r.get(callback=on_result, propagate=propagate, **kwargs)
-            except (socket.timeout, TimeoutError) as exc:
-                waiting_for = self.missing_results(r)
-                self.speaker.beep()
-                marker(
-                    'Still waiting for {0}/{1}: [{2}]: {3!r}'.format(
-                        len(r) - len(received), len(r),
-                        truncate(', '.join(waiting_for)), exc), '!',
-                )
-                self.fbi.diag(waiting_for)
-            except self.connerrors as exc:
-                self.speaker.beep()
-                marker('join: connection lost: {0!r}'.format(exc), '!')
-        raise StopSuite('Test failed: Missing task results')
-
-    def dump_progress(self):
-        return pstatus(self.progress) if self.progress else 'No test running'
-
-
-_creation_counter = count(0)
-
-
-def testcase(*groups, **kwargs):
-    if not groups:
-        raise ValueError('@testcase requires at least one group name')
-
-    def _mark_as_case(fun):
-        fun.__testgroup__ = groups
-        fun.__testsort__ = next(_creation_counter)
-        fun.__iterations__ = kwargs.get('iterations')
-        return fun
-
-    return _mark_as_case
-
-
-def _is_descriptor(obj, attr):
-    try:
-        cattr = getattr(obj.__class__, attr)
-    except AttributeError:
-        pass
-    else:
-        return not inspect.ismethod(cattr) and hasattr(cattr, '__get__')
-    return False
-
-
-class Suite(BaseSuite):
-
-    @testcase('all', 'green', 'redis', iterations=1)
-    def chain(self):
-        c = add.s(4, 4) | add.s(8) | add.s(16)
-        assert_equal(self.join(c()), 32)
-
-    @testcase('all', 'green', 'redis', iterations=1)
-    def chaincomplex(self):
-        c = (
-            add.s(2, 2) | (
-                add.s(4) | add.s(8) | add.s(16)
-            ) |
-            group(add.s(i) for i in range(4))
-        )
-        res = c()
-        assert_equal(res.get(), [32, 33, 34, 35])
-
-    @testcase('all', 'green', 'redis', iterations=1)
-    def parentids_chain(self, num=248):
-        c = chain(ids.si(i) for i in range(num))
-        c.freeze()
-        res = c()
-        res.get(timeout=5)
-        self.assert_ids(res, num - 1)
-
-    @testcase('all', 'green', 'redis', iterations=1)
-    def parentids_group(self):
-        g = ids.si(1) | ids.si(2) | group(ids.si(i) for i in range(2, 50))
-        res = g()
-        expected_root_id = res.parent.parent.id
-        expected_parent_id = res.parent.id
-        values = res.get(timeout=5)
-
-        for i, r in enumerate(values):
-            root_id, parent_id, value = r
-            assert_equal(root_id, expected_root_id)
-            assert_equal(parent_id, expected_parent_id)
-            assert_equal(value, i + 2)
-
-    def assert_ids(self, res, size):
-        i, root = size, res
-        while root.parent:
-            root = root.parent
-        node = res
-        while node:
-            root_id, parent_id, value = node.get(timeout=5)
-            assert_equal(value, i)
-            assert_equal(root_id, root.id)
-            if node.parent:
-                assert_equal(parent_id, node.parent.id)
-            node = node.parent
-            i -= 1
-
-    @testcase('redis', iterations=1)
-    def parentids_chord(self):
-        self.assert_parentids_chord()
-        self.assert_parentids_chord(uuid(), uuid())
-
-    def assert_parentids_chord(self, base_root=None, base_parent=None):
-        g = (
-            ids.si(1) |
-            ids.si(2) |
-            group(ids.si(i) for i in range(3, 50)) |
-            collect_ids.s(i=50) |
-            ids.si(51)
-        )
-        g.freeze(root_id=base_root, parent_id=base_parent)
-        res = g.apply_async(root_id=base_root, parent_id=base_parent)
-        expected_root_id = base_root or res.parent.parent.parent.id
-
-        root_id, parent_id, value = res.get(timeout=5)
-        assert_equal(value, 51)
-        assert_equal(root_id, expected_root_id)
-        assert_equal(parent_id, res.parent.id)
-
-        prev, (root_id, parent_id, value) = res.parent.get(timeout=5)
-        assert_equal(value, 50)
-        assert_equal(root_id, expected_root_id)
-        assert_equal(parent_id, res.parent.parent.id)
-
-        for i, p in enumerate(prev):
-            root_id, parent_id, value = p
-            assert_equal(root_id, expected_root_id)
-            assert_equal(parent_id, res.parent.parent.id)
-
-        root_id, parent_id, value = res.parent.parent.get(timeout=5)
-        assert_equal(value, 2)
-        assert_equal(parent_id, res.parent.parent.parent.id)
-        assert_equal(root_id, expected_root_id)
-
-        root_id, parent_id, value = res.parent.parent.parent.get(timeout=5)
-        assert_equal(value, 1)
-        assert_equal(root_id, expected_root_id)
-        assert_equal(parent_id, base_parent)
-
-    @testcase('all', 'green')
-    def manyshort(self):
-        self.join(group(add.s(i, i) for i in range(1000))(),
-                  timeout=10, propagate=True)
-
-    @testcase('all', 'green', iterations=1)
-    def unicodetask(self):
-        self.join(group(print_unicode.s() for _ in range(5))(),
-                  timeout=1, propagate=True)
-
-    @testcase('all')
-    def always_timeout(self):
-        self.join(
-            group(sleeping.s(1).set(time_limit=0.1)
-                  for _ in range(100))(),
-            timeout=10, propagate=True,
-        )
-
-    @testcase('all')
-    def termbysig(self):
-        self._evil_groupmember(kill)
-
-    @testcase('green')
-    def group_with_exit(self):
-        self._evil_groupmember(exiting)
-
-    @testcase('all')
-    def timelimits(self):
-        self._evil_groupmember(sleeping, 2, time_limit=1)
-
-    @testcase('all')
-    def timelimits_soft(self):
-        self._evil_groupmember(sleeping_ignore_limits, 2,
-                               soft_time_limit=1, time_limit=1.1)
-
-    @testcase('all')
-    def alwayskilled(self):
-        g = group(kill.s() for _ in range(10))
-        self.join(g(), timeout=10)
-
-    @testcase('all', 'green')
-    def alwaysexits(self):
-        g = group(exiting.s() for _ in range(10))
-        self.join(g(), timeout=10)
-
-    def _evil_groupmember(self, evil_t, *eargs, **opts):
-        g1 = group(add.s(2, 2).set(**opts), evil_t.s(*eargs).set(**opts),
-                   add.s(4, 4).set(**opts), add.s(8, 8).set(**opts))
-        g2 = group(add.s(3, 3).set(**opts), add.s(5, 5).set(**opts),
-                   evil_t.s(*eargs).set(**opts), add.s(7, 7).set(**opts))
-        self.join(g1(), timeout=10)
-        self.join(g2(), timeout=10)
-
-    @testcase('all', 'green')
-    def bigtasksbigvalue(self):
-        g = group(any_returning.s(BIG, sleep=0.3) for i in range(8))
-        r = g()
-        try:
-            self.join(r, timeout=10)
-        finally:
-            # very big values so remove results from backend
-            try:
-                r.forget()
-            except NotImplementedError:
-                pass
-
-    @testcase('all', 'green')
-    def bigtasks(self, wait=None):
-        self._revoketerm(wait, False, False, BIG)
-
-    @testcase('all', 'green')
-    def smalltasks(self, wait=None):
-        self._revoketerm(wait, False, False, SMALL)
-
-    @testcase('all')
-    def revoketermfast(self, wait=None):
-        self._revoketerm(wait, True, False, SMALL)
-
-    @testcase('all')
-    def revoketermslow(self, wait=5):
-        self._revoketerm(wait, True, True, BIG)
-
-    def _revoketerm(self, wait=None, terminate=True,
-                    joindelay=True, data=BIG):
-        g = group(any_.s(data, sleep=wait) for i in range(8))
-        r = g()
-        if terminate:
-            if joindelay:
-                sleep(random.choice(range(4)))
-            r.revoke(terminate=True)
-        self.join(r, timeout=10)

funtests/stress/stress/templates.py

@@ -1,157 +0,0 @@
-from __future__ import absolute_import, unicode_literals
-
-import celery
-import os
-
-from functools import partial
-
-from celery.five import items
-from kombu import Queue
-from kombu.utils import symbol_by_name
-
-CSTRESS_TRANS = os.environ.get('CSTRESS_TRANS', False)
-default_queue = 'c.stress.trans' if CSTRESS_TRANS else 'c.stress'
-CSTRESS_QUEUE = os.environ.get('CSTRESS_QUEUE_NAME', default_queue)
-
-templates = {}
-
-IS_CELERY_4 = celery.VERSION[0] >= 4
-
-
-def template(name=None):
-
-    def _register(cls):
-        templates[name or cls.__name__] = '.'.join([__name__, cls.__name__])
-        return cls
-    return _register
-
-
-if IS_CELERY_4:
-
-    def use_template(app, template='default'):
-        template = template.split(',')
-
-        # mixin the rest of the templates when the config is needed
-        @app.on_after_configure.connect(weak=False)
-        def load_template(sender, source, **kwargs):
-            mixin_templates(template[1:], source)
-
-        app.config_from_object(templates[template[0]])
-else:
-
-    def use_template(app, template='default'):  # noqa
-        template = template.split(',')
-        app.after_configure = partial(mixin_templates, template[1:])
-        app.config_from_object(templates[template[0]])
-
-
-def mixin_templates(templates, conf):
-    return [mixin_template(template, conf) for template in templates]
-
-
-def mixin_template(template, conf):
-    cls = symbol_by_name(templates[template])
-    conf.update(dict(
-        (k, v) for k, v in items(vars(cls))
-        if not k.startswith('_')
-    ))
-
-
-def template_names():
-    return ', '.join(templates)
-
-
-@template()
-class default(object):
-    CELERY_ACCEPT_CONTENT = ['json']
-    BROKER_URL = os.environ.get('CSTRESS_BROKER', 'pyamqp://')
-    BROKER_HEARTBEAT = 30
-    CELERY_RESULT_BACKEND = os.environ.get('CSTRESS_BACKEND', 'rpc://')
-    CELERY_RESULT_SERIALIZER = 'json'
-    CELERY_RESULT_PERSISTENT = True
-    CELERY_RESULT_EXPIRES = 300
-    CELERY_MAX_CACHED_RESULTS = 100
-    CELERY_DEFAULT_QUEUE = CSTRESS_QUEUE
-    CELERY_TASK_QUEUES = [
-        Queue(CSTRESS_QUEUE,
-              durable=not CSTRESS_TRANS,
-              no_ack=CSTRESS_TRANS),
-    ]
-    CELERY_TASK_SERIALIZER = 'json'
-    CELERY_TASK_PUBLISH_RETRY_POLICY = {
-        'max_retries': 100,
-        'interval_max': 2,
-        'interval_step': 0.1,
-    }
-    CELERY_TASK_PROTOCOL = 2
-    if CSTRESS_TRANS:
-        CELERY_DEFAULT_DELIVERY_MODE = 1
-    CELERYD_PREFETCH_MULTIPLIER = int(os.environ.get('CSTRESS_PREFETCH', 10))
-
-
-@template()
-class redis(default):
-    BROKER_URL = os.environ.get('CSTRESS_BROKER', 'redis://')
-    BROKER_TRANSPORT_OPTIONS = {
-        'fanout_prefix': True,
-        'fanout_patterns': True,
-    }
-    CELERY_RESULT_BACKEND = os.environ.get('CSTRESS_BACKEND', 'redis://')
-
-
-@template()
-class redistore(default):
-    CELERY_RESULT_BACKEND = 'redis://'
-
-
-@template()
-class acks_late(default):
-    CELERY_ACKS_LATE = True
-
-
-@template()
-class pickle(default):
-    CELERY_ACCEPT_CONTENT = ['pickle', 'json']
-    CELERY_TASK_SERIALIZER = 'pickle'
-    CELERY_RESULT_SERIALIZER = 'pickle'
-
-
-@template()
-class confirms(default):
-    BROKER_URL = 'pyamqp://'
-    BROKER_TRANSPORT_OPTIONS = {'confirm_publish': True}
-
-
-@template()
-class events(default):
-    CELERY_SEND_EVENTS = True
-    CELERY_SEND_TASK_SENT_EVENT = True
-
-
-@template()
-class execv(default):
-    CELERYD_FORCE_EXECV = True
-
-
-@template()
-class sqs(default):
-    BROKER_URL = 'sqs://'
-    BROKER_TRANSPORT_OPTIONS = {
-        'region': os.environ.get('AWS_REGION', 'us-east-1'),
-    }
-
-
-@template()
-class proto1(default):
-    CELERY_TASK_PROTOCOL = 1
-
-
-@template()
-class vagrant1(default):
-    BROKER_URL = 'pyamqp://testing:t3s71ng@192.168.33.123//testing'
-
-
-@template()
-class vagrant1_redis(redis):
-    BROKER_URL = 'redis://192.168.33.123'
-    CELERY_RESULT_BACKEND = 'redis://192.168.33.123'

requirements/pkgutils.txt

@@ -4,3 +4,4 @@ flake8>=2.5.4
 flakeplus>=1.1
 tox>=2.3.1
 sphinx2rst>=1.0
+cyanide>=1.0