I have a Celery app (4.4.6) that uses dataclasses. Since JSON can't serialize/deserialize dataclasses, I have forced the use of pickle throughout (I'm aware of the risk, but I think it's mitigated in the way the app is deployed). However, I am getting errors from within the app from kombu, saying TypeError: Object of type ResourceGroup is not JSON serializable. Everything else is working, so in general, it has to be using pickle OK, but in this one case it isn't. However nothing in the stacktrace that comes with the exception mentions my code. The structure of the software is tasks that create other tasks dynamically as they perform discovery, using delay(). Almost all the tasks are passed this ResourceGroup object, and they are running fine, except one (I think, judging by the frequency of these errors, and the logging I am getting for completing tasks).
This is how I'm configuring Celery in my worker. Is there some other setting I need to set to really really make it use pickle in all situations? (alternatively, is there a JSON serializer/deserializer that can reconstitute dataclasses?)
class CeleryConfig:
task_serializer = 'pickle'
result_serializer = 'pickle'
event_serializer = 'pickle'
accept_content = ['pickle']
result_accept_content = ['pickle']
app = Celery('collector',
backend='redis://' + Config.REDIS_HOST,
broker='redis://' + Config.REDIS_HOST,
include=['tasks'])
app.config_from_object(CeleryConfig)
Update:
Here's a full example of one of the exceptions:
[2020-07-23 05:02:10,621: DEBUG/MainProcess] pidbox received method active() [reply_to:{'exchange': 'reply.celery.pidbox', 'routing_key': 'cde2e89b-bb81-3c19-8491-b57b072e5f29'} ticket:6c2cc493-0e2d-4a85-821b-350bdc4bceeb]
[2020-07-23 05:02:10,621: ERROR/MainProcess] Control command error: EncodeError(TypeError('Object of type ResourceGroup is not JSON serializable'))
Traceback (most recent call last):
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 50, in _reraise_errors
yield
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 221, in dumps
payload = encoder(data)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 69, in dumps
return _dumps(s, cls=cls or _default_encoder,
File "/usr/local/lib/python3.8/json/__init__.py", line 234, in dumps
return cls(
File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode
chunks = self.iterencode(o, _one_shot=True)
File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode
return _iterencode(o, 0)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 59, in default
return super(JSONEncoder, self).default(o)
File "/usr/local/lib/python3.8/json/encoder.py", line 179, in default
raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type ResourceGroup is not JSON serializable
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/site-packages/celery/worker/pidbox.py", line 46, in on_message
self.node.handle_message(body, message)
File "/usr/local/lib/python3.8/site-packages/kombu/pidbox.py", line 145, in handle_message
return self.dispatch(**body)
File "/usr/local/lib/python3.8/site-packages/kombu/pidbox.py", line 112, in dispatch
self.reply({self.hostname: reply},
File "/usr/local/lib/python3.8/site-packages/kombu/pidbox.py", line 149, in reply
self.mailbox._publish_reply(data, exchange, routing_key, ticket,
File "/usr/local/lib/python3.8/site-packages/kombu/pidbox.py", line 280, in _publish_reply
producer.publish(
File "/usr/local/lib/python3.8/site-packages/kombu/messaging.py", line 167, in publish
body, content_type, content_encoding = self._prepare(
File "/usr/local/lib/python3.8/site-packages/kombu/messaging.py", line 252, in _prepare
body) = dumps(body, serializer=serializer)
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 221, in dumps
payload = encoder(data)
File "/usr/local/lib/python3.8/contextlib.py", line 131, in __exit__
self.gen.throw(type, value, traceback)
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 54, in _reraise_errors
reraise(wrapper, wrapper(exc), sys.exc_info()[2])
File "/usr/local/lib/python3.8/site-packages/vine/five.py", line 194, in reraise
raise value.with_traceback(tb)
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 50, in _reraise_errors
yield
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 221, in dumps
payload = encoder(data)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 69, in dumps
return _dumps(s, cls=cls or _default_encoder,
File "/usr/local/lib/python3.8/json/__init__.py", line 234, in dumps
return cls(
File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode
chunks = self.iterencode(o, _one_shot=True)
File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode
return _iterencode(o, 0)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 59, in default
return super(JSONEncoder, self).default(o)
File "/usr/local/lib/python3.8/json/encoder.py", line 179, in default
raise TypeError(f'Object of type {o.__class__.__name__} '
kombu.exceptions.EncodeError: Object of type ResourceGroup is not JSON serializable
I agree that ResourceGroup isn't JSON serializable. I just don't understand why it's trying! :-)
update2:
I got the celery worker running with concurrency=1 in the debugger, and now with a breakpoint in the JSON serializer (which I should never hit) I get a stop. From reading back inside kombu, the first mention of serializer in the stack trace seems to be from Node.reply() in pidbox.py where self.mailbox.serializer is None (and defaults to JSON because of that). I don't see where that Mailbox object is originally created though.
update3:
The Control object appears to always create a Mailbox that will accept and send JSON, regardless of any config:
"""Worker remote control client."""
Mailbox = Mailbox
def __init__(self, app=None):
self.app = app
self.mailbox = self.Mailbox(
app.conf.control_exchange,
type='fanout',
accept=['json'],
producer_pool=lazy(lambda: self.app.amqp.producer_pool),
queue_ttl=app.conf.control_queue_ttl,
reply_queue_ttl=app.conf.control_queue_ttl,
queue_expires=app.conf.control_queue_expires,
reply_queue_expires=app.conf.control_queue_expires,
)