celery purge revocation control command

purge_revoked.py

from celery.worker.control import control_command

from celery.worker import state as worker_state


@control_command(
    args=[('n', float)],
    signature='[N=0]',  # <- used for help on the command-line.
)
def purge_revoked(state, n=0):
    if n:
        worker_state.revoked.purge(n)
    else:
        worker_state.revoked.clear()
    return {'ok': 'purged all revoked task info.'}

@control_command(
    args=[('n', int)],
    signature='[N={}]'.format(worker_state.REVOKES_MAX),  # <- used for help on the command-line.
)
def revokes_max(state, n=worker_state.REVOKES_MAX):
    worker_state.revoked.maxlen = n
    worker_state.revoked.purge()
    return {'ok': 'updated revoked task max length.'}


@control_command(
    args=[('n', int)],
    signature='[N={}]'.format(worker_state.REVOKE_EXPIRES),  # <- used for help on the command-line.
)
def revokes_expires(state, n=worker_state.REVOKE_EXPIRES):
    worker_state.revoked.expires = n
    worker_state.revoked.purge()
    return {'ok': 'updated revoked task expiration.'}

@jheld Hi, sorry to revive this old thread. May I just ask, how did you call these remote control commands? Was it periodically from an external scheduled process?

I think primarily it was part of a script within continuous deployment. So we'd run it I think before deploying new code. We also tended to opt for scaling the workers to 0 before starting a roll-out as well. A large part of the issue being addressed with these commands is that new workers coming up would simply add to the memory issue, so as we deployed new code, if we had scaled all workers entirely down before adding any new, then I think the general issue goes away. But yes, you could still add this at any cadence of your comfort.