Kombu: python exceptions are not json serializable.

[jayendra13]

I have some celery tasks, which may raise exception, if required. The problem is python Exception object is not json-serialization, yet though yaml can serialize python Exception objects but I don't have control to change this configuration for celery.

Can we in celery(more specifically in kombu) internally handle, serialization of exception object ?

Here is a snippet which will generate error.

from kombu import Connection

ex = None
try:
    a = 1/0
except ZeroDivisionError, e:
    ex = e
print ex

conn = Connection('amqp://guest:guest@localhost:5672//')
simple_queue = conn.SimpleQueue('simple_queue')

# No error as expected
simple_queue.put('Hello')

# No error as expected
simple_queue.put(str(ex))

# json cannot serliaze exception
# hence celery gives error
simple_queue.put(ex)

Here is error which I get.

EncodeError                               Traceback (most recent call last)
<ipython-input-22-9667148d81ba> in <module>()
----> 1 sq.put(ex)

/usr/local/lib/python2.7/dist-packages/kombu/simple.pyc in put(self, message, serializer, headers, compression, routing_key, **kwargs)
     70                               headers=headers,
     71                               compression=compression,
---> 72                               **kwargs)
     73 
     74     def clear(self):

/usr/local/lib/python2.7/dist-packages/kombu/messaging.pyc in publish(self, body, routing_key, delivery_mode, mandatory, immediate, priority, content_type, content_encoding, serializer, headers, compression, exchange, retry, retry_policy, declare, expiration, **properties)
    163         body, content_type, content_encoding = self._prepare(
    164             body, serializer, content_type, content_encoding,
--> 165             compression, headers)
    166 
    167         publish = self._publish

/usr/local/lib/python2.7/dist-packages/kombu/messaging.pyc in _prepare(self, body, serializer, content_type, content_encoding, compression, headers)
    239             serializer = serializer or self.serializer
    240             (content_type, content_encoding,
--> 241              body) = dumps(body, serializer=serializer)
    242         else:
    243             # If the programmer doesn't want us to serialize,

/usr/local/lib/python2.7/dist-packages/kombu/serialization.pyc in dumps(self, data, serializer)
    162 
    163         with _reraise_errors(EncodeError):
--> 164             payload = encoder(data)
    165         return content_type, content_encoding, payload
    166     encode = dumps  # XXX compat

/usr/lib/python2.7/contextlib.pyc in __exit__(self, type, value, traceback)
     33                 value = type()
     34             try:
---> 35                 self.gen.throw(type, value, traceback)
     36                 raise RuntimeError("generator didn't stop after throw()")
     37             except StopIteration, exc:

/usr/local/lib/python2.7/dist-packages/kombu/serialization.pyc in _reraise_errors(wrapper, include, exclude)
     57         raise
     58     except include as exc:
---> 59         reraise(wrapper, wrapper(exc), sys.exc_info()[2])
     60 
     61 

/usr/local/lib/python2.7/dist-packages/kombu/serialization.pyc in _reraise_errors(wrapper, include, exclude)
     53                     include=(Exception, ), exclude=(SerializerNotInstalled, )):
     54     try:
---> 55         yield
     56     except exclude:
     57         raise

/usr/local/lib/python2.7/dist-packages/kombu/serialization.pyc in dumps(self, data, serializer)
    162 
    163         with _reraise_errors(EncodeError):
--> 164             payload = encoder(data)
    165         return content_type, content_encoding, payload
    166     encode = dumps  # XXX compat

/usr/local/lib/python2.7/dist-packages/anyjson/__init__.pyc in dumps(value)
    139     def dumps(value):
    140         """Deserialize JSON-encoded object to a Python object."""
--> 141         return implementation.dumps(value)
    142     serialize = dumps

/usr/local/lib/python2.7/dist-packages/anyjson/__init__.pyc in dumps(self, data)
     85         TypeError if the object could not be serialized."""
     86         try:
---> 87             return self._encode(data)
     88         except self._encode_error, exc:
     89             raise TypeError, TypeError(*exc.args), sys.exc_info()[2]

/usr/local/lib/python2.7/dist-packages/simplejson-3.8.1-py2.7-linux-x86_64.egg/simplejson/__init__.pyc in dumps(obj, skipkeys, ensure_ascii, check_circular, allow_nan, cls, indent, separators, encoding, default, use_decimal, namedtuple_as_object, tuple_as_array, bigint_as_string, sort_keys, item_sort_key, for_json, ignore_nan, int_as_string_bitcount, iterable_as_array, **kw)
    378         and not kw
    379     ):
--> 380         return _default_encoder.encode(obj)
    381     if cls is None:
    382         cls = JSONEncoder

/usr/local/lib/python2.7/dist-packages/simplejson-3.8.1-py2.7-linux-x86_64.egg/simplejson/encoder.pyc in encode(self, o)
    273         # exceptions aren't as detailed.  The list call should be roughly
    274         # equivalent to the PySequence_Fast that ''.join() would do.
--> 275         chunks = self.iterencode(o, _one_shot=True)
    276         if not isinstance(chunks, (list, tuple)):
    277             chunks = list(chunks)

/usr/local/lib/python2.7/dist-packages/simplejson-3.8.1-py2.7-linux-x86_64.egg/simplejson/encoder.pyc in iterencode(self, o, _one_shot)
    355                 self.iterable_as_array, Decimal=decimal.Decimal)
    356         try:
--> 357             return _iterencode(o, 0)
    358         finally:
    359             key_memo.clear()

/usr/local/lib/python2.7/dist-packages/simplejson-3.8.1-py2.7-linux-x86_64.egg/simplejson/encoder.pyc in default(self, o)
    250 
    251         """
--> 252         raise TypeError(repr(o) + " is not JSON serializable")
    253 
    254     def encode(self, o):

[malinoff]

If you follow some rules, celery will be able to serialize your exception objects. Could you please show us your tasks?

[jayendra13]

My task looks like this. It would be perfectly fine if it returns after raising exception. The problem is now my execution of task fails with this not serialization error.

def pre_build(my_var, tickers=[]):

    # some junk code here
    try:
        assert len(my_var) == 0, "there should not be any h5 files in {} yet".format(distdir)
    except AssertionError, ex:
        logging.info("Target dir is not empty")
        return

    # some junk code here too

some version info if required
python=2.7
kombu==3.0.28
anyjson==0.3.3
jsonschema==2.5.1
simplejson==3.8.1
ujson==1.34

[malinoff]

This is not valid python code. Please, show us _exact_ code you have (with stripped private data, of course).

[jayendra13]

OK, it will take much time, to strip off my code (and its my lunch time too ;)). @malinoff Can you point me the rules you were talking about in your previous comment ?

[malinoff]

You're likely to raise a custom exception instance which isn't subclassed from Exception.

[jayendra13]

This is the exact location where, I am getting the error, and reason is clear that python json library can't handle python exception class.

23 @app.task
24 def execute_task(task_loc, task_name, *args, **kwargs):
25     module = __import__(task_loc, globals(), locals(), [task_name], 0)
26     try:
27         return getattr(module, task_name)(*args, **kwargs)
28     except Exception as ex:
29         traceback.print_exc()
30         return ex

Here was the similar problem for set
https://github.com/celery/kombu/issues/177

I am suggesting that can't we handle the exceptions in same manner as set ?
like this

import yaml
#while serilization
#where json is trying to
#serialize the object
if isinstance(object, Exception):
    serialized = yaml.dump(object)

#while deserilization
#where json is trying to 
#de-serialize the object 
try:
    object = json.loads(serialized)
except:
    object = yaml.load(serialized)

I am not aware of the consequences of it, but this approach will not throw errors, further if json fails to serialise objects.

[ask]

Kombu doesn't care about the content of your messages, celery has a specific task protocol and tools to deal with exceptions.

Celery task results can definitely deal with exceptions using json, in the case where it don't it would be a bug and should be reported to the celery issue tracker!

[humitos]

Hi! I'm sorry to comment in a closed issue but I just found that I'm experimenting a similar problem:

kombu.exceptions.EncodeError: Object of type 'Exception' is not JSON serializable

I'm getting this when calling self.update_state(state='FAILURE', meta={'error': Exception('Error')} inside the after_return method.

I've been reading the documentation and dealing with this error for a couple of hours now and I'm not sure why this is happening.

The original code is at https://github.com/rtfd/readthedocs.org/blob/master/readthedocs/core/utils/tasks/public.py#L83

Let me know if I can provide more information. Just in case, I tried this isolated example and the exception is properly serialized and returned: https://github.com/celery/celery/issues/2518#issue-59154540

[humitos]

Based on that example I mentioned in my previous comment, I wrote this one that reproduces the issue:

# jsonresults.py
#
#

# python -m celery --app=jsonresults:app worker -l INFO


# $ python
# Python 3.6.4 (default, Mar 15 2018, 09:44:44)
# [GCC 7.3.0] on linux
# Type "help", "copyright", "credits" or "license" for more information.
# >>> import jsonresults ; result = jsonresults.MyTask().delay()
# >>> result
# <AsyncResult: 75387cda-4998-41da-a50d-87080726d999>
# >>> result.info
# EncodeError("Object of type 'ValueError' is not JSON serializable",)
# >>>


from celery.app.base import Celery
from celery import Task

CELERY_RESULT_SERIALIZER = 'json'
BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'


app = Celery(config_source=__name__)


class MyTask(Task):
    name = 'MyTask'

    def run(self):
        raise ValueError('go away')

    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        info = {}
        if status == 'FAILURE':
            info['error'] = retval
        self.update_state(state=status, meta=info)

app.tasks.register(MyTask())

[georgepsarakis]

@humitos you can use self.backend.store_result, passing the exception instance, as seen here. This is a different use case, so please open a separate issue, so that we have better tracking and knowledge base. Let me know if this works!

[humitos]

@georgepsarakis thank you! I just opened a new issue at https://github.com/celery/kombu/issues/868

[jheld]

is there a place where I can post a similar issue? My team uses elasticsearch for some of the data and some tasks fail because of ConnectionTimeout from elasticsearch. When this happens, the worker half-crashed (stopped consuming tasks but was still an active process). We have in the past had to turn off CELERY_TASK_STORE_ERRORS_EVEN_IF_IGNORED, but this is still an issue when not ignoring results in general so we've also had to add ignore_result=True to the affected tasks. Ideally we shouldn't have to do that (and it may limit our ability in general to use the celery canvas objects effectively if the need were to arise).

Please advise and if there is a way to fix it and you would accept the PR, I'd be happy to contribute back.

[georgepsarakis]

@jheld I think what you describe belongs to a separate issue.

[mac10046]

Hey guys I guess this should work good enough for any Exception which is not serializable

    except Exception as e:
        import json
        json.dumps(e.__dict__)

This worked for me atleast.

[jheld]

Do we agree that the snippet provided above is worthwhile as an official fix/suggestion?