Removes autoscale option completely

README.rst

@@ -121,7 +121,7 @@ Celery is...
 
     Almost every part of *Celery* can be extended or used on its own,
     Custom pool implementations, serializers, compression schemes, logging,
-    schedulers, consumers, producers, autoscalers, broker transports and much more.
+    schedulers, consumers, producers, broker transports and much more.
 
 It supports...
 ============

celery/app/control.py

@@ -290,15 +290,6 @@ class Control(object):
         """
         return self.broadcast('pool_shrink', {'n': n}, destination, **kwargs)
 
-    def autoscale(self, max, min, destination=None, **kwargs):
-        """Change worker(s) autoscale setting.
-
-        Supports the same arguments as :meth:`broadcast`.
-
-        """
-        return self.broadcast(
-            'autoscale', {'max': max, 'min': min}, destination, **kwargs)
-
     def broadcast(self, command, arguments=None, destination=None,
                   connection=None, reply=False, timeout=1, limit=None,
                   callback=None, channel=None, **extra_kwargs):

celery/app/defaults.py

@@ -266,7 +266,6 @@ NAMESPACES = Namespace(
     worker=Namespace(
         __old__=OLD_NS_WORKER,
         agent=Option(None, type='string'),
-        autoscaler=Option('celery.worker.autoscale:Autoscaler'),
         autoreloader=Option('celery.worker.autoreload:Autoreloader'),
         concurrency=Option(0, type='int'),
         consumer=Option('celery.worker.consumer:Consumer', type='string'),

celery/apps/worker.py

@@ -189,9 +189,6 @@ class Worker(WorkController):
             if loader.startswith('celery.loaders'):  # pragma: no cover
                 loader = loader[14:]
             appr += ' ({0})'.format(loader)
-        if self.autoscale:
-            max, min = self.autoscale
-            concurrency = '{{min={0}, max={1}}}'.format(min, max)
         pool = self.pool_cls
         if not isinstance(pool, string_t):
             pool = pool.__module__

celery/bin/celery.py

@@ -702,7 +702,6 @@ class control(_RemoteControl):
             1.0, 'tell worker(s) to modify the rate limit for a task type'),
         'time_limit': (
             1.0, 'tell worker(s) to modify the time limit for a task type.'),
-        'autoscale': (1.0, 'change autoscale settings'),
         'pool_grow': (1.0, 'start more pool processes'),
         'pool_shrink': (1.0, 'use less pool processes'),
     }
@@ -718,10 +717,6 @@ class control(_RemoteControl):
         """[N=1]"""
         return self.call(method, int(n), **kwargs)
 
-    def autoscale(self, method, max=None, min=None, **kwargs):
-        """[max] [min]"""
-        return self.call(method, int(max), int(min), **kwargs)
-
     def rate_limit(self, method, task_name, rate_limit, **kwargs):
         """<task_name> <rate_limit> (e.g. 5/s | 5/m | 5/h)>"""
         return self.call(method, task_name, rate_limit, **kwargs)

celery/bin/worker.py

@@ -119,15 +119,6 @@ The :program:`celery worker` command (previously known as ``celeryd``)
     completed and the child process will be replaced afterwards.
     Default: no limit.
 
-.. cmdoption:: --autoscale
-
-    Enable autoscaling by providing
-    max_concurrency, min_concurrency. Example::
-
-        --autoscale=10,3
-
-    (always keep 3 processes, but grow to 10 if necessary)
-
 .. cmdoption:: --autoreload
 
     Enable auto-reloading.
@@ -209,7 +200,6 @@ class worker(Command):
         celery worker -A proj --concurrency=4
         celery worker -A proj --concurrency=1000 -P eventlet
 
-        celery worker --autoscale=10,0
     """
     doc = __MODULE_DOC__  # parse help from this too
     namespace = 'worker'
@@ -332,7 +322,6 @@ class worker(Command):
         parser.add_option_group(qopts)
 
         fopts = OptionGroup(parser, 'Features')
-        fopts.add_option('--autoscale')
         fopts.add_option('--autoreload', action='store_true')
         fopts.add_option(
             '--without-gossip', action='store_true', default=False,

celery/tests/app/test_app.py

@@ -297,7 +297,7 @@ class test_App(AppCase):
                 task_default_delivery_mode=63,
                 worker_agent='foo:Barz',
                 CELERYD_CONSUMER='foo:Fooz',
-                CELERYD_AUTOSCALER='foo:Xuzzy',
+                CELERYD_POOL='foo:Xuzzy',
             )
             with self.assertRaises(ImproperlyConfigured):
                 self.assertEqual(app.conf.worker_consumer, 'foo:Fooz')
@@ -310,11 +310,11 @@ class test_App(AppCase):
                 worker_agent='foo:Barz',
                 CELERYD_CONSUMER='foo:Fooz',
                 worker_consumer='foo:Fooz',
-                CELERYD_AUTOSCALER='foo:Xuzzy',
-                worker_autoscaler='foo:Xuzzy'
+                CELERYD_POOL='foo:Xuzzy',
+                worker_pool='foo:Xuzzy'
             )
             self.assertEqual(app.conf.task_always_eager, 4)
-            self.assertEqual(app.conf.worker_autoscaler, 'foo:Xuzzy')
+            self.assertEqual(app.conf.worker_pool, 'foo:Xuzzy')
 
     def test_pending_configuration__setdefault(self):
         with self.Celery(broker='foo://bar') as app:

celery/tests/bin/test_celery.py

@@ -541,11 +541,6 @@ class test_control(AppCase):
         i.pool_shrink('pool_shrink', n=2)
         i.call.assert_called_with('pool_shrink', 2)
 
-    def test_autoscale(self):
-        i = self.control(True)
-        i.autoscale('autoscale', max=3, min=2)
-        i.call.assert_called_with('autoscale', 3, 2)
-
     def test_rate_limit(self):
         i = self.control(True)
         i.rate_limit('rate_limit', 'proj.add', '1/s')

celery/tests/bin/test_celeryd_detach.py

@@ -91,7 +91,7 @@ class test_PartialOptionParser(AppCase):
 
 
 class test_Command(AppCase):
-    argv = ['--autoscale=10,2', '-c', '1',
+    argv = ['--foobar=10,2', '-c', '1',
             '--logfile=/var/log', '-lDEBUG',
             '--', '.disable_rate_limits=1']
 
@@ -99,7 +99,7 @@ class test_Command(AppCase):
         x = detached_celeryd(app=self.app)
         o, v, l = x.parse_options('cd', self.argv)
         self.assertEqual(o.logfile, '/var/log')
-        self.assertEqual(l, ['--autoscale=10,2', '-c', '1',
+        self.assertEqual(l, ['--foobar=10,2', '-c', '1',
                              '-lDEBUG', '--logfile=/var/log',
                              '--pidfile=celeryd.pid'])
         x.parse_options('cd', [])  # no args

celery/tests/bin/test_worker.py

@@ -150,8 +150,6 @@ class test_Worker(WorkerAppCase):
         self.assertTrue(worker.startup_info())
         worker.loglevel = logging.INFO
         self.assertTrue(worker.startup_info())
-        worker.autoscale = 13, 10
-        self.assertTrue(worker.startup_info())
 
         prev_loader = self.app.loader
         worker = self.Worker(app=self.app, queues='foo,bar,baz,xuzzy,do,re,mi')
@@ -221,14 +219,6 @@ class test_Worker(WorkerAppCase):
             app.amqp.queues['image'],
         )
 
-    @mock.stdouts
-    def test_autoscale_argument(self, stdout, stderr):
-        worker1 = self.Worker(app=self.app, autoscale='10,3')
-        self.assertListEqual(worker1.autoscale, [10, 3])
-        worker2 = self.Worker(app=self.app, autoscale='10')
-        self.assertListEqual(worker2.autoscale, [10, 0])
-        self.assert_no_logging_side_effect()
-
     def test_include_argument(self):
         worker1 = self.Worker(app=self.app, include='os')
         self.assertListEqual(worker1.include, ['os'])

celery/tests/worker/test_autoscale.py

@@ -1,217 +0,0 @@
-from __future__ import absolute_import, unicode_literals
-
-import sys
-
-from celery.concurrency.base import BasePool
-from celery.five import monotonic
-from celery.worker import state
-from celery.worker import autoscale
-from celery.utils.objects import Bunch
-
-from celery.tests.case import AppCase, Mock, mock, patch
-
-
-class MockPool(BasePool):
-    shrink_raises_exception = False
-    shrink_raises_ValueError = False
-
-    def __init__(self, *args, **kwargs):
-        super(MockPool, self).__init__(*args, **kwargs)
-        self._pool = Bunch(_processes=self.limit)
-
-    def grow(self, n=1):
-        self._pool._processes += n
-
-    def shrink(self, n=1):
-        if self.shrink_raises_exception:
-            raise KeyError('foo')
-        if self.shrink_raises_ValueError:
-            raise ValueError('foo')
-        self._pool._processes -= n
-
-    @property
-    def num_processes(self):
-        return self._pool._processes
-
-
-class test_WorkerComponent(AppCase):
-
-    def test_register_with_event_loop(self):
-        parent = Mock(name='parent')
-        parent.autoscale = True
-        parent.consumer.on_task_message = set()
-        w = autoscale.WorkerComponent(parent)
-        self.assertIsNone(parent.autoscaler)
-        self.assertTrue(w.enabled)
-
-        hub = Mock(name='hub')
-        w.create(parent)
-        w.register_with_event_loop(parent, hub)
-        self.assertIn(
-            parent.autoscaler.maybe_scale,
-            parent.consumer.on_task_message,
-        )
-        hub.call_repeatedly.assert_called_with(
-            parent.autoscaler.keepalive, parent.autoscaler.maybe_scale,
-        )
-
-        parent.hub = hub
-        hub.on_init = []
-        w.instantiate = Mock()
-        w.register_with_event_loop(parent, Mock(name='loop'))
-        self.assertTrue(parent.consumer.on_task_message)
-
-
-class test_Autoscaler(AppCase):
-
-    def setup(self):
-        self.pool = MockPool(3)
-
-    def test_stop(self):
-
-        class Scaler(autoscale.Autoscaler):
-            alive = True
-            joined = False
-
-            def is_alive(self):
-                return self.alive
-
-            def join(self, timeout=None):
-                self.joined = True
-
-        worker = Mock(name='worker')
-        x = Scaler(self.pool, 10, 3, worker=worker)
-        x._is_stopped.set()
-        x.stop()
-        self.assertTrue(x.joined)
-        x.joined = False
-        x.alive = False
-        x.stop()
-        self.assertFalse(x.joined)
-
-    @mock.sleepdeprived(module=autoscale)
-    def test_body(self):
-        worker = Mock(name='worker')
-        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
-        x.body()
-        self.assertEqual(x.pool.num_processes, 3)
-        _keep = [Mock(name='req{0}'.format(i)) for i in range(20)]
-        [state.task_reserved(m) for m in _keep]
-        x.body()
-        x.body()
-        self.assertEqual(x.pool.num_processes, 10)
-        worker.consumer._update_prefetch_count.assert_called()
-        state.reserved_requests.clear()
-        x.body()
-        self.assertEqual(x.pool.num_processes, 10)
-        x._last_scale_up = monotonic() - 10000
-        x.body()
-        self.assertEqual(x.pool.num_processes, 3)
-        worker.consumer._update_prefetch_count.assert_called()
-
-    def test_run(self):
-
-        class Scaler(autoscale.Autoscaler):
-            scale_called = False
-
-            def body(self):
-                self.scale_called = True
-                self._is_shutdown.set()
-
-        worker = Mock(name='worker')
-        x = Scaler(self.pool, 10, 3, worker=worker)
-        x.run()
-        self.assertTrue(x._is_shutdown.isSet())
-        self.assertTrue(x._is_stopped.isSet())
-        self.assertTrue(x.scale_called)
-
-    def test_shrink_raises_exception(self):
-        worker = Mock(name='worker')
-        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
-        x.scale_up(3)
-        x.pool.shrink_raises_exception = True
-        x._shrink(1)
-
-    @patch('celery.worker.autoscale.debug')
-    def test_shrink_raises_ValueError(self, debug):
-        worker = Mock(name='worker')
-        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
-        x.scale_up(3)
-        x._last_scale_up = monotonic() - 10000
-        x.pool.shrink_raises_ValueError = True
-        x.scale_down(1)
-        self.assertTrue(debug.call_count)
-
-    def test_update_and_force(self):
-        worker = Mock(name='worker')
-        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
-        self.assertEqual(x.processes, 3)
-        x.force_scale_up(5)
-        self.assertEqual(x.processes, 8)
-        x.update(5, None)
-        self.assertEqual(x.processes, 5)
-        x.force_scale_down(3)
-        self.assertEqual(x.processes, 2)
-        x.update(None, 3)
-        self.assertEqual(x.processes, 3)
-        x.force_scale_down(1000)
-        self.assertEqual(x.min_concurrency, 0)
-        self.assertEqual(x.processes, 0)
-        x.force_scale_up(1000)
-        x.min_concurrency = 1
-        x.force_scale_down(1)
-
-        x.update(max=300, min=10)
-        x.update(max=300, min=2)
-        x.update(max=None, min=None)
-
-    def test_info(self):
-        worker = Mock(name='worker')
-        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
-        info = x.info()
-        self.assertEqual(info['max'], 10)
-        self.assertEqual(info['min'], 3)
-        self.assertEqual(info['current'], 3)
-
-    @patch('os._exit')
-    def test_thread_crash(self, _exit):
-
-        class _Autoscaler(autoscale.Autoscaler):
-
-            def body(self):
-                self._is_shutdown.set()
-                raise OSError('foo')
-        worker = Mock(name='worker')
-        x = _Autoscaler(self.pool, 10, 3, worker=worker)
-
-        stderr = Mock()
-        p, sys.stderr = sys.stderr, stderr
-        try:
-            x.run()
-        finally:
-            sys.stderr = p
-        _exit.assert_called_with(1)
-        self.assertTrue(stderr.write.call_count)
-
-    @mock.sleepdeprived(module=autoscale)
-    def test_no_negative_scale(self):
-        total_num_processes = []
-        worker = Mock(name='worker')
-        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
-        x.body()  # the body func scales up or down
-
-        _keep = [Mock(name='req{0}'.format(i)) for i in range(35)]
-        for req in _keep:
-            state.task_reserved(req)
-            x.body()
-            total_num_processes.append(self.pool.num_processes)
-
-        for req in _keep:
-            state.task_ready(req)
-            x.body()
-            total_num_processes.append(self.pool.num_processes)
-
-        self. assertTrue(
-            all(x.min_concurrency <= i <= x.max_concurrency
-                for i in total_num_processes)
-        )

celery/tests/worker/test_control.py

@@ -26,7 +26,6 @@ hostname = socket.gethostname()
 
 
 class WorkController(object):
-    autoscaler = None
 
     def stats(self):
         return {'total': worker_state.total_count}
@@ -336,11 +335,6 @@ class test_ControlPanel(AppCase):
 
         panel.state.consumer = Mock()
         panel.state.consumer.controller = Mock()
-        sc = panel.state.consumer.controller.autoscaler = Mock()
-        panel.handle('pool_grow')
-        sc.force_scale_up.assert_called()
-        panel.handle('pool_shrink')
-        sc.force_scale_down.assert_called()
 
     def test_add__cancel_consumer(self):
 
@@ -501,21 +495,6 @@ class test_ControlPanel(AppCase):
         finally:
             worker_state.task_ready(request)
 
-    def test_autoscale(self):
-        self.panel.state.consumer = Mock()
-        self.panel.state.consumer.controller = Mock()
-        sc = self.panel.state.consumer.controller.autoscaler = Mock()
-        sc.update.return_value = 10, 2
-        m = {'method': 'autoscale',
-             'destination': hostname,
-             'arguments': {'max': '10', 'min': '2'}}
-        r = self.panel.handle_message(m, None)
-        self.assertIn('ok', r)
-
-        self.panel.state.consumer.controller.autoscaler = None
-        r = self.panel.handle_message(m, None)
-        self.assertIn('error', r)
-
     def test_ping(self):
         m = {'method': 'ping',
              'destination': hostname}

celery/tests/worker/test_worker.py

@@ -775,13 +775,6 @@ class test_WorkController(AppCase):
         self.assertTrue(worker.beat)
         self.assertIn(worker.beat, [w.obj for w in worker.steps])
 
-    def test_with_autoscaler(self):
-        worker = self.create_worker(
-            autoscale=[10, 3], send_events=False,
-            timer_cls='celery.utils.timer2.Timer',
-        )
-        self.assertTrue(worker.autoscaler)
-
     def test_dont_stop_or_terminate(self):
         worker = self.app.WorkController(concurrency=1, loglevel=0)
         worker.stop()

celery/worker/__init__.py

@@ -82,7 +82,6 @@ class WorkController(object):
             'celery.worker.components:Timer',
             'celery.worker.components:StateDB',
             'celery.worker.components:Consumer',
-            'celery.worker.autoscale:WorkerComponent',
             'celery.worker.autoreload:WorkerComponent',
         }
 
@@ -342,7 +341,7 @@ class WorkController(object):
     def setup_defaults(self, concurrency=None, loglevel='WARN', logfile=None,
                        send_events=None, pool_cls=None, consumer_cls=None,
                        timer_cls=None, timer_precision=None,
-                       autoscaler_cls=None, autoreloader_cls=None,
+                       autoreloader_cls=None,
                        pool_putlocks=None, pool_restarts=None,
                        force_execv=None, state_db=None,
                        schedule_filename=None, scheduler_cls=None,
@@ -362,7 +361,6 @@ class WorkController(object):
         self.timer_precision = either(
             'worker_timer_precision', timer_precision,
         )
-        self.autoscaler_cls = either('worker_autoscaler', autoscaler_cls)
         self.autoreloader_cls = either('worker_autoreloader', autoreloader_cls)
         self.pool_putlocks = either('worker_pool_putlocks', pool_putlocks)
         self.pool_restarts = either('worker_pool_restarts', pool_restarts)

celery/worker/autoscale.py

@@ -1,161 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-    celery.worker.autoscale
-    ~~~~~~~~~~~~~~~~~~~~~~~
-
-    This module implements the internal thread responsible
-    for growing and shrinking the pool according to the
-    current autoscale settings.
-
-    The autoscale thread is only enabled if
-    the :option:`celery worker --autoscale` option is used.
-
-"""
-from __future__ import absolute_import, unicode_literals
-
-import os
-import threading
-
-from time import sleep
-
-from kombu.async.semaphore import DummyLock
-
-from celery import bootsteps
-from celery.five import monotonic
-from celery.utils.log import get_logger
-from celery.utils.threads import bgThread
-
-from . import state
-from .components import Pool
-
-__all__ = ['Autoscaler', 'WorkerComponent']
-
-logger = get_logger(__name__)
-debug, info, error = logger.debug, logger.info, logger.error
-
-AUTOSCALE_KEEPALIVE = float(os.environ.get('AUTOSCALE_KEEPALIVE', 30))
-
-
-class WorkerComponent(bootsteps.StartStopStep):
-    label = 'Autoscaler'
-    conditional = True
-    requires = (Pool,)
-
-    def __init__(self, w, **kwargs):
-        self.enabled = w.autoscale
-        w.autoscaler = None
-
-    def create(self, w):
-        scaler = w.autoscaler = self.instantiate(
-            w.autoscaler_cls,
-            w.pool, w.max_concurrency, w.min_concurrency,
-            worker=w, mutex=DummyLock() if w.use_eventloop else None,
-        )
-        return scaler if not w.use_eventloop else None
-
-    def register_with_event_loop(self, w, hub):
-        w.consumer.on_task_message.add(w.autoscaler.maybe_scale)
-        hub.call_repeatedly(
-            w.autoscaler.keepalive, w.autoscaler.maybe_scale,
-        )
-
-
-class Autoscaler(bgThread):
-
-    def __init__(self, pool, max_concurrency,
-                 min_concurrency=0, worker=None,
-                 keepalive=AUTOSCALE_KEEPALIVE, mutex=None):
-        super(Autoscaler, self).__init__()
-        self.pool = pool
-        self.mutex = mutex or threading.Lock()
-        self.max_concurrency = max_concurrency
-        self.min_concurrency = min_concurrency
-        self.keepalive = keepalive
-        self._last_scale_up = None
-        self.worker = worker
-
-        assert self.keepalive, 'cannot scale down too fast.'
-
-    def body(self):
-        with self.mutex:
-            self.maybe_scale()
-        sleep(1.0)
-
-    def _maybe_scale(self, req=None):
-        procs = self.processes
-        cur = min(self.qty, self.max_concurrency)
-        if cur > procs:
-            self.scale_up(cur - procs)
-            return True
-        cur = max(self.qty, self.min_concurrency)
-        if cur < procs:
-            self.scale_down(procs - cur)
-            return True
-
-    def maybe_scale(self, req=None):
-        if self._maybe_scale(req):
-            self.pool.maintain_pool()
-
-    def update(self, max=None, min=None):
-        with self.mutex:
-            if max is not None:
-                if max < self.processes:
-                    self._shrink(self.processes - max)
-                self.max_concurrency = max
-            if min is not None:
-                if min > self.processes:
-                    self._grow(min - self.processes)
-                self.min_concurrency = min
-            return self.max_concurrency, self.min_concurrency
-
-    def force_scale_up(self, n):
-        with self.mutex:
-            new = self.processes + n
-            if new > self.max_concurrency:
-                self.max_concurrency = new
-            self._grow(n)
-
-    def force_scale_down(self, n):
-        with self.mutex:
-            new = self.processes - n
-            if new < self.min_concurrency:
-                self.min_concurrency = max(new, 0)
-            self._shrink(min(n, self.processes))
-
-    def scale_up(self, n):
-        self._last_scale_up = monotonic()
-        return self._grow(n)
-
-    def scale_down(self, n):
-        if self._last_scale_up and (
-                monotonic() - self._last_scale_up > self.keepalive):
-            return self._shrink(n)
-
-    def _grow(self, n):
-        info('Scaling up %s processes.', n)
-        self.pool.grow(n)
-        self.worker.consumer._update_prefetch_count(n)
-
-    def _shrink(self, n):
-        info('Scaling down %s processes.', n)
-        try:
-            self.pool.shrink(n)
-        except ValueError:
-            debug("Autoscaler won't scale down: all processes busy.")
-        except Exception as exc:
-            error('Autoscaler: scale_down: %r', exc, exc_info=True)
-        self.worker.consumer._update_prefetch_count(-n)
-
-    def info(self):
-        return {'max': self.max_concurrency,
-                'min': self.min_concurrency,
-                'current': self.processes,
-                'qty': self.qty}
-
-    @property
-    def qty(self):
-        return len(state.reserved_requests)
-
-    @property
-    def processes(self):
-        return self.pool.num_processes

celery/worker/components.py

@@ -108,7 +108,6 @@ class Pool(bootsteps.StartStopStep):
 
     Adds attributes:
 
-        * autoscale
         * pool
         * max_concurrency
         * min_concurrency
@@ -116,18 +115,12 @@ class Pool(bootsteps.StartStopStep):
     """
     requires = (Hub,)
 
-    def __init__(self, w, autoscale=None, autoreload=None,
+    def __init__(self, w, autoreload=None,
                  no_execv=False, optimization=None, **kwargs):
-        if isinstance(autoscale, string_t):
-            max_c, _, min_c = autoscale.partition(',')
-            autoscale = [int(max_c), min_c and int(min_c) or 0]
-        w.autoscale = autoscale
         w.pool = None
         w.max_concurrency = None
         w.min_concurrency = w.concurrency
         w.no_execv = no_execv
-        if w.autoscale:
-            w.max_concurrency, w.min_concurrency = w.autoscale
         self.autoreload_enabled = autoreload
         self.optimization = optimization
 

celery/worker/consumer/consumer.py

@@ -246,8 +246,9 @@ class Consumer(object):
         .. note::
 
             Currently pool grow operations will end up with an offset
-            of +1 if the initial size of the pool was 0 (e.g.
-            :option:`--autoscale=1,0 <celery worker --autoscale>`).
+            of +1 if the initial size of the pool was 0 (which could
+            be the case with old deprecated autoscale option, may consider
+            removing this now that it's no longer supported).
 
         """
         num_processes = self.pool.num_processes

celery/worker/control.py

@@ -341,21 +341,15 @@ def memdump(state, samples=10, **kwargs):  # pragma: no cover
 
 @Panel.register
 def pool_grow(state, n=1, **kwargs):
-    if state.consumer.controller.autoscaler:
-        state.consumer.controller.autoscaler.force_scale_up(n)
-    else:
-        state.consumer.pool.grow(n)
-        state.consumer._update_prefetch_count(n)
+    state.consumer.pool.grow(n)
+    state.consumer._update_prefetch_count(n)
     return ok('pool will grow')
 
 
 @Panel.register
 def pool_shrink(state, n=1, **kwargs):
-    if state.consumer.controller.autoscaler:
-        state.consumer.controller.autoscaler.force_scale_down(n)
-    else:
-        state.consumer.pool.shrink(n)
-        state.consumer._update_prefetch_count(-n)
+    state.consumer.pool.shrink(n)
+    state.consumer._update_prefetch_count(-n)
     return ok('pool will shrink')
 
 
@@ -368,15 +362,6 @@ def pool_restart(state, modules=None, reload=False, reloader=None, **kwargs):
         raise ValueError('Pool restarts not enabled')
 
 
-@Panel.register
-def autoscale(state, max=None, min=None):
-    autoscaler = state.consumer.controller.autoscaler
-    if autoscaler:
-        max_, min_ = autoscaler.update(max, min)
-        return ok('autoscale now max={0} min={1}'.format(max_, min_))
-    raise ValueError('Autoscale not enabled')
-
-
 @Panel.register
 def shutdown(state, msg='Got shutdown from remote', **kwargs):
     logger.warning(msg)

docs/configuration.rst

@@ -142,7 +142,6 @@ rush in moving to the new settings format.
 ``CELERYD_TASK_TIME_LIMIT``            :setting:`task_time_limit`
 ``CELERY_TRACK_STARTED``               :setting:`task_track_started`
 ``CELERYD_AGENT``                      :setting:`worker_agent`
-``CELERYD_AUTOSCALER``                 :setting:`worker_autoscaler`
 ``CELERYD_AUTORELAODER``               :setting:`worker_autoreloader`
 ``CELERYD_CONCURRENCY``                :setting:`worker_concurrency`
 ``CELERYD_CONSUMER``                   :setting:`worker_consumer`
@@ -2231,17 +2230,6 @@ If enabled the worker pool can be restarted using the
 
 Disabled by default.
 
-.. setting:: worker_autoscaler
-
-``worker_autoscaler``
-~~~~~~~~~~~~~~~~~~~~~
-
-.. versionadded:: 2.2
-
-Name of the autoscaler class to use.
-
-Default is ``celery.worker.autoscale:Autoscaler``.
-
 .. setting:: worker_autoreloader
 
 ``worker_autoreloader``

docs/getting-started/introduction.rst

@@ -108,7 +108,7 @@ Celery is…
 
         Almost every part of *Celery* can be extended or used on its own,
         Custom pool implementations, serializers, compression schemes, logging,
-        schedulers, consumers, producers, autoscalers, broker transports and much more.
+        schedulers, consumers, producers, broker transports and much more.
 
 
 .. topic:: It supports

docs/includes/introduction.txt

@@ -113,7 +113,7 @@ Celery is…
 
     Almost every part of *Celery* can be extended or used on its own,
     Custom pool implementations, serializers, compression schemes, logging,
-    schedulers, consumers, producers, autoscalers, broker transports and much more.
+    schedulers, consumers, producers, broker transports and much more.
 
 It supports…
 ============

docs/internals/reference/celery.worker.autoscale.rst

@@ -1,11 +0,0 @@
-========================================
- ``celery.worker.autoscale``
-========================================
-
-.. contents::
-    :local:
-.. currentmodule:: celery.worker.autoscale
-
-.. automodule:: celery.worker.autoscale
-    :members:
-    :undoc-members:

docs/internals/reference/index.rst

@@ -14,7 +14,6 @@
     celery.worker.control
     celery.worker.pidbox
     celery.worker.autoreload
-    celery.worker.autoscale
     celery.concurrency
     celery.concurrency.solo
     celery.concurrency.prefork

docs/userguide/extending.rst

@@ -112,9 +112,12 @@ Worker
 ======
 
 The Worker is the first blueprint to start, and with it starts major components like
-the event loop, processing pool, the timer, and also optional components
-like the autoscaler.  When the worker is fully started it will continue
-to the Consumer blueprint.
+the event loop, processing pool, and the timer used for ETA tasks and other
+timed events.
+
+When the worker is fully started it will continue to the Consumer blueprint,
+which sets up how tasks are to be executed, connects to the broker and starts
+the message consumers.
 
 The :class:`~celery.worker.WorkController` is the core worker implementation,
 and contains several methods and attributes that you can use in your bootstep.
@@ -202,22 +205,6 @@ Attributes
         class WorkerStep(bootsteps.StartStopStep):
             requires = ('celery.worker.components:Statedb',)
 
-.. _extending-worker-autoscaler:
-
-.. attribute:: autoscaler
-
-    :class:`~celery.worker.autoscaler.Autoscaler` used to automatically grow
-    and shrink the number of processes in the pool.
-
-    This is only defined if the ``autoscale`` argument is enabled.
-
-    Your worker bootstep must require the `Autoscaler` bootstep to use this:
-
-    .. code-block:: python
-
-        class WorkerStep(bootsteps.StartStopStep):
-            requires = ('celery.worker.autoscaler:Autoscaler',)
-
 .. _extending-worker-autoreloader:
 
 .. attribute:: autoreloader
@@ -666,8 +653,7 @@ will show us more information about the boot process:
     [2013-05-29 16:18:20,511: DEBUG/MainProcess] | Worker: Building graph...
     <celery.apps.worker.Worker object at 0x101ad8410> is in init
     [2013-05-29 16:18:20,511: DEBUG/MainProcess] | Worker: New boot order:
-        {Hub, Pool, Autoreloader, Timer, StateDB,
-         Autoscaler, InfoStep, Beat, Consumer}
+        {Hub, Pool, Autoreloader, Timer, StateDB, InfoStep, Beat, Consumer}
     [2013-05-29 16:18:20,514: DEBUG/MainProcess] | Consumer: Preparing bootsteps.
     [2013-05-29 16:18:20,514: DEBUG/MainProcess] | Consumer: Building graph...
     <celery.worker.consumer.Consumer object at 0x101c2d8d0> is in init

docs/userguide/monitoring.rst

@@ -229,7 +229,7 @@ Features
 
     - View worker status and statistics
     - Shutdown and restart worker instances
-    - Control worker pool size and autoscale settings
+    - Control worker pool size
     - View and modify the queues a worker instance consumes from
     - View currently running tasks
     - View scheduled tasks (ETA/countdown)
@@ -245,7 +245,6 @@ Features
     - Restart worker’s pool
     - Grow worker’s pool
     - Shrink worker’s pool
-    - Autoscale worker pool
     - Start consuming from a queue
     - Stop consuming from a queue
     - List tasks

docs/userguide/workers.rst

@@ -177,7 +177,7 @@ filename depending on the process that will eventually need to open the file.
 This can be used to specify one log file per child process.
 
 Note that the numbers will stay within the process limit even if processes
-exit or if autoscale/``maxtasksperchild``/time limits are used.  I.e. the number
+exit or if ``maxtasksperchild``/time limits are used.  I.e. the number
 is the *process index* not the process count or pid.
 
 * ``%i`` - Pool process index or 0 if MainProcess.
@@ -553,37 +553,6 @@ The option can be set using the workers
 :option:`--maxmemperchild <celery worker --maxmemperchild>` argument
 or using the :setting:`worker_max_memory_per_child` setting.
 
-.. _worker-autoscaling:
-
-Autoscaling
-===========
-
-.. versionadded:: 2.2
-
-:pool support: *prefork*, *gevent*
-
-The *autoscaler* component is used to dynamically resize the pool
-based on load:
-
-- The autoscaler adds more pool processes when there is work to do,
-    - and starts removing processes when the workload is low.
-
-It's enabled by the :option:`--autoscale <celery worker --autoscale>` option,
-which needs two numbers: the maximum and minimum number of pool processes:
-
-.. code-block:: text
-
-        --autoscale=AUTOSCALE
-             Enable autoscaling by providing
-             max_concurrency,min_concurrency.  Example:
-               --autoscale=10,3 (always keep 3 processes, but grow to
-              10 if necessary).
-
-You can also define your own rules for the autoscaler by subclassing
-:class:`~celery.worker.autoscaler.Autoscaler`.
-Some ideas for metrics include load average or the amount of memory available.
-You can specify a custom autoscaler with the :setting:`worker_autoscaler` setting.
-
 .. _worker-queues:
 
 Queues

extra/bash-completion/celery.bash

@@ -75,7 +75,7 @@ _celery()
         COMPREPLY=( $(compgen -W '--concurrency= --pool= --purge --logfile=
         --loglevel= --hostname= --beat --schedule= --scheduler= --statedb= --events
         --time-limit= --soft-time-limit= --maxtasksperchild= --queues=
-        --include= --pidfile= --autoscale= --autoreload --no-execv $fargs' -- ${cur} ) )
+        --include= --pidfile= --autoreload --no-execv $fargs' -- ${cur} ) )
         return 0
         ;;
     inspect)
@@ -84,7 +84,7 @@ _celery()
         return 0
         ;;
     control)
-        COMPREPLY=( $(compgen -W 'add_consumer autoscale cancel_consumer
+        COMPREPLY=( $(compgen -W 'add_consumer cancel_consumer
         disable_events enable_events pool_grow pool_shrink
         rate_limit time_limit --help $controlargs $fargs' -- ${cur}) )
         return 0

extra/zsh-completion/celery.zsh

@@ -55,7 +55,6 @@ case "$words[1]" in
     '(-Q --queues=)'{-Q,--queues=}'[List of queues to enable for this worker, separated by comma. By default all configured queues are enabled.]' \
     '(-I --include=)'{-I,--include=}'[Comma separated list of additional modules to import.]' \
     '(--pidfile=)--pidfile=[Optional file used to store the process pid.]' \
-    '(--autoscale=)--autoscale=[Enable autoscaling by providing max_concurrency, min_concurrency.]' \
     '(--autoreload)--autoreload[Enable autoreloading.]' \
     '(--no-execv)--no-execv[Don"t do execv after multiprocessing child fork.]'
     compadd -a ifargs
@@ -76,7 +75,6 @@ case "$words[1]" in
     control)
     _values -s \
     'add_consumer[tell worker(s) to start consuming a queue]' \
-    'autoscale[change autoscale settings]' \
     'cancel_consumer[tell worker(s) to stop consuming a queue]' \
     'disable_events[tell worker(s) to disable events]' \
     'enable_events[tell worker(s) to enable events]' \