Pulp

This issue is triaged as "Issue B".
@dralley will file a new issue for "Issue A".

Applied in changeset pulp|881a5fb9fdaf9813d0dbb576ab7ca3c7b3dc8476.

I accidentally pushed this to pulp/2.14-dev instead of my own Github. Apologies.

PR is here: https://github.com/pulp/pulp/pull/3137

Summary:¶

Restarting a worker that is currently executing a task will leave that workers in a broken state. This issue can reproduced on both Celery 3.1.x and Celery 4.x, but only while using Qpid as a broker. I was not able to reproduce this issue while using RabbitMQ as a broker, using either version of Celery. I was also not able to reproduce this issue on versions of Pulp prior to 2.13. The means of shutting down the workers also does not appear to matter, e.g. "systemctl restart" and "pkill -9 celery; prestart" both work the same.

Reproduction Steps:¶

1. Start pulp
2. Begin a task (e.g. sync)
3. While the task is running, restart the pulp worker running the task
4. After the worker has restarted, begin another task
5. Observe that the tasks are perpetually stuck in waiting

(Exact steps)

1. prestart
2. pulp-admin rpm repo sync run --repo-id zoo

(While the task is still running, in another terminal)

3a. sudo systemctl restart pulp_workers
OR
3b. pkill -9 celery; prestart

4. pulp-admin rpm repo sync run --repo-id zoo
5. Observe that the task is perpetually stuck in waiting

Symptoms:¶

Only the worker which was running a task at the time of the worker restart is rendered frozen. Future work assigned to this worker will not be executed - this can cause it to seem as though Pulp is entirely frozen, because work is typically assigned to worker zero. Other workers are fine.

From the CLI:

  [vagrant@pulp2 ~]$ pulp-admin rpm repo sync run --repo-id zoo
  +----------------------------------------------------------------------+
                       Synchronizing Repository [zoo]
  +----------------------------------------------------------------------+

  This command may be exited via ctrl+c without affecting the request.

  Downloading metadata...
  [\]
  ... completed

  Downloading repository content...
  [-]
  [==================================================] 100%
  RPMs:       32/32 items
  Delta RPMs: 0/0 items

  Task Canceled

  [vagrant@pulp2 ~]$ pulp-admin rpm repo sync run --repo-id zoo
  +----------------------------------------------------------------------+
                       Synchronizing Repository [zoo]
  +----------------------------------------------------------------------+

  This command may be exited via ctrl+c without affecting the request.

  [\]
  Waiting to begin...

We can see that Pulp worker is not consuming work from it's associated dedicated queue - the messages sent to the queue by dispatching a new task are not being read out of the queue by the worker.

State of queues before running any tasks

  queue                                               dur  autoDel  excl  msg   msgIn  msgOut  bytes  bytesIn  bytesOut  cons  bind
  ===================================================================================================================================
  reserved_resource_worker-0@pulp2.dev.dq             Y    Y                 0     0      0       0      0        0         1     2
  resource_manager                                    Y                      0    36     36       0   43.1k    43.1k        1     2

State of queues after restarting worker mid-task

  queue                                               dur  autoDel  excl  msg   msgIn  msgOut  bytes  bytesIn  bytesOut  cons  bind
  ===================================================================================================================================
  reserved_resource_worker-0@pulp2.dev.dq             Y    Y                 0     0      0       0      0        0         1     2
  resource_manager                                    Y                      0    40     40       0   47.8k    47.8k        1     2

State of queues after dispatching a new task to the frozen worker

  queue                                               dur  autoDel  excl  msg   msgIn  msgOut  bytes  bytesIn  bytesOut  cons  bind
  ===================================================================================================================================
  reserved_resource_worker-0@pulp2.dev.dq             Y    Y                 2     2      0    1.93k  1.93k       0         1     2
  resource_manager                                    Y                      0    41     41       0   49.0k    49.0k        1     2

If you restart the worker immediately after restarting it mid-task, without assigning any new work to the queue, it will restart cleanly and will continue on accepting work normally.

How to "fix":¶

List all of the tasks in waiting with 'pulp-admin tasks list', cancel them all with 'pulp-admin tasks list cancel', and then restart the worker

• If you cancel the new task without restarting the worker and then issue a new task, that task will become hung just the same
• If you restart the worker without cancelling a current hung task, the task will be marked cancelled on reboot of the worker, but future work dispatched to the worker will remain hung
• If you do not assign new work to the worker, then restart the worker, it will continue accepting work normally

Origin of this issue:¶

A change was introduced in Pulp 2.13 which used of Celery's Bootsteps feature as a means for workers to write their own heartbeats to the database. This appears to have triggered the conditions to make this bug emergent.

That PR is here: https://github.com/pulp/pulp/pull/2922

After this change, when the function we registered to be executed on the celeryd_after_setup signal is run, the call to _delete_worker() -> cancel() -> controller.revoke() appears cause the worker to become frozen, whereas previously it had worked fine.

https://github.com/pulp/pulp/blob/master/server/pulp/server/async/tasks.py#L663
https://github.com/pulp/pulp/blob/master/server/pulp/server/async/app.py#L158

The exact reason that using bootsteps causes the call to revoke() to put the worker in a bad state is still undetermined.

Mitigation:¶

My current pull request [0] to Pulp can mitigate this issue by registering a new terminate() function on the Celery consumer (called as part of the shutdown process of the worker) which calls _delete_worker() itself. This only fixes the case where the workers are shutdown "kindly", but this is going to be the most common case, and the fix definitely does work 100% for that case.

[0] https://github.com/pulp/pulp/pull/3137/files

Pulp already code which was supposed to accomplish this using the worker_shutdown signal [1], but upon testing I found that while the signal was being called correctly, the qpid.messaging thread had already been stopped at that point, and the resulting traceback prevented the code from reaching the portion of _delete_worker() where the Pulp task is marked cancelled. Therefore, when the worker restarts, it attempts to clean up those workers, calling revoke() and getting frozen up.

[1] https://github.com/pulp/pulp/blob/master/server/pulp/server/async/app.py#L171

This approach fixes the issue in the case where the workers are shut down "nicely" aka with SIGTERM. An expanded fix also solves the "pkill" case by moving the pulp worker cleanup and initialization code from the celeryd_after_setup signal handler to the start() method on the Consumer bootstep. However, I have no early idea WHY this works, or why having that code celeryd_after_setup broke when we started using bootsteps in the first place. So I'm not comfortable pushing that expanded fix until those questions are answered.

Debugging:¶

Attached is the output of running "thread apply all py-bt" on GDB coredumps of the parent and child Celery processes in 4 states.

nowork_* => Clean processes before any work whatsoever has been dispatched
postsync_* => Processes after work has been dispatched normally, without restarting the workers mid-execution
restarted_* => Processes after the pulp workers have been restarted mid-execution of a task
hung_* => Processes after being restarted and having new work dispatched, which are now hung

All 4 parent dumps are identical. The two "clean" child dumps, nowork_child.txt and postsync_child.txt, are also identical to each other, as are the two "dirty" child dumps, hung_child.txt and restarted_child.txt. So, for the sake of comparing differences between the dumps, we only need to look at two - which may as well be nowork_child.txt and hung_child.txt

The aforementioned "full workaround" I found by tinkering, but which I don't understand how it works, is here:

https://github.com/pulp/pulp/compare/master...dralley:2835-hack-fix

I consider this to be hacky... this doesn't address why the introduction of bootsteps broke previously-functional code within the celeryd_after_setup signal handler.

I've attached two additional logs, where I started some minimal celery workers which are structurally identical to the pulp workers, and watched their startup sequences.

test_worker1 (similar to the pre-2.13 pulp worker code)

from celery import Celery
from celery.signals import celeryd_after_setup

@celeryd_after_setup.connect
def test_signal(sender, instance, **kwargs):
    print('{0!r} celeryd_after_setup signal fired'.format(sender))

app = Celery(broker='amqp://')

test_worker2 (similar to the 2.13+ pulp_worker code)

from celery import Celery
from celery import bootsteps
from celery.signals import celeryd_after_setup

class Reproducer(bootsteps.StartStopStep):
    def __init__(self, parent, **kwargs):
        print('{0!r} is in init'.format(parent))

    def start(self, worker):
        print('{0!r} is starting up'.format(worker))

        self.timer_ref = worker.timer.call_repeatedly(
            5,
            self.do_work,
            (worker, ),
            priority=10,
        )

    def do_work(self, worker):
        print('{0!r} heartbeat'.format(worker))

    def stop(self, parent):
        print('{0!r} is stopping'.format(parent))

    def shutdown(self, parent):
        print('{0!r} is shutting down'.format(parent))

@celeryd_after_setup.connect
def test_signal(sender, instance, **kwargs):
    print('{0!r} celeryd_after_setup signal fired'.format(sender))

app = Celery(broker='amqp://')
app.steps['consumer'].add(Reproducer)

There are differences in the logs, but I don't see those differences as being meaningful, since they all occur after the point that we care about.

Applied in changeset pulp|aa7a1de219b02bb0aa5a3674a72d77a653ee968f.

Tested this by following the steps from #15