Handling Exceptions in Nameko

googlegroup
1

Hi, I'm trying to use Nameko for the creation of a simple notification
center that sends emails consuming from several queues in RabbitMQ.

Our whole architecture is service based and we have a system to write the
exceptions of every service in our database that are queued in a different
queue in Rabbit too. Is there a way to handle the exceptions that raise
during a worker job? I notice that they are logged and the service is still
running. But what I want is to publish a message in the Error queue when an
exception raises besides the log. I've tried with sys.excepthook and raised
an exception on purpose but the exception was catched before it could be
handled by the excepthook. Any ideas?

Thanks in advance!

Alfonso

2

Hi Alfonso,

This is very easy to do by hooking in to the "worker lifecycle". Just add a
DependencyProvider that implements worker_result. Example below:

from nameko.rpc import rpc
from nameko.extensions import DependencyProvider

class ErrorHandler(DependencyProvider):

    def worker_result(self, worker_ctx, res, exc_info):
        if exc_info is None:
            return # nothing to do

        exc_type, exc, tb = exc_info
        # do whatever you need with the exception here
        print(exc)

class Service:
    name = "service"

    handler = ErrorHandler()

    @rpc
    def method(self):
        raise Exception("something went wrong")

Running service:

(nameko)Matts-13-inch-Macbook-Pro:tmp mattbennett$ nameko run service
starting services: service
Connected to amqp://guest:**@127.0.0.1:5672//
error handling worker <WorkerContext [service.method] at 0x10e6aa790>:
something went wrong
Traceback (most recent call last):
  File "/Users/mattbennett/Code/nameko/nameko/containers.py", line 405, in
_run_worker
    result = method(*worker_ctx.args, **worker_ctx.kwargs)
  File "./service.py", line 23, in method
    raise Exception("something went wrong")
Exception: something went wrong
something went wrong # <<< this is print(exc)

Use the shell to trigger an exception:

(nameko)Matts-13-inch-Macbook-Pro:nameko mattbennett$ nameko shell
Nameko Python 2.7.10 (default, Jul 6 2015, 15:19:48)
[GCC 4.2.1 Compatible Apple LLVM 6.1.0 (clang-602.0.53)] shell on darwin
Broker: pyamqp://guest:guest@localhost

n.rpc.service.method()

Traceback (most recent call last):
  File "<input>", line 1, in <module>
    n.rpc.service.method()
  File "/Users/mattbennett/Code/nameko/nameko/rpc.py", line 395, in __call__
    return reply.result()
  File "/Users/mattbennett/Code/nameko/nameko/rpc.py", line 381, in result
    raise deserialize(error)
RemoteError: Exception something went wrong

···

On Thursday, April 6, 2017 at 12:33:53 PM UTC+1, alfo...@gmail.com wrote:

Hi, I'm trying to use Nameko for the creation of a simple notification
center that sends emails consuming from several queues in RabbitMQ.

Our whole architecture is service based and we have a system to write the
exceptions of every service in our database that are queued in a different
queue in Rabbit too. Is there a way to handle the exceptions that raise
during a worker job? I notice that they are logged and the service is still
running. But what I want is to publish a message in the Error queue when an
exception raises besides the log. I've tried with sys.excepthook and raised
an exception on purpose but the exception was catched before it could be
handled by the excepthook. Any ideas?

Thanks in advance!

Alfonso

3

That's it! Thank you Matt!

···

On Thursday, April 6, 2017 at 1:33:53 PM UTC+2, alfo...@gmail.com wrote:

Hi, I'm trying to use Nameko for the creation of a simple notification
center that sends emails consuming from several queues in RabbitMQ.

Our whole architecture is service based and we have a system to write the
exceptions of every service in our database that are queued in a different
queue in Rabbit too. Is there a way to handle the exceptions that raise
during a worker job? I notice that they are logged and the service is still
running. But what I want is to publish a message in the Error queue when an
exception raises besides the log. I've tried with sys.excepthook and raised
an exception on purpose but the exception was catched before it could be
handled by the excepthook. Any ideas?

Thanks in advance!

Alfonso

4

Hi Matt, I'm having some trouble with this. The idea is that when I catch
the exception with the worker_result, I want to publish a message in the
exchange called ErrorExchange. I have this code:

import traceback

from kombu.messaging import Exchange
from nameko.extensions import DependencyProvider
from nameko.messaging import Publisher

error_exchange = Exchange("ErrorExchange", type='fanout')

class ErrorHandler(DependencyProvider):
    
    p = Publisher(exchange=error_exchange)

    def worker_result(self, worker_ctx, res, exc_info):
        
        if exc_info is None:
            return # nothing to do

        exc_type, exc, tb = exc_info
        msg = { "message": str(exc_info), "stack":
traceback.format_tb(tb), "type": str(exc_type) }
        self.p(msg) #Publish message in the exchange

But when I try to run, I get this error:

(cn) am@VdiUB-02:~/Projects/utils/cn$ nameko run notificationcenter
--broker amqp://guest:guest@localhost
Traceback (most recent call last):
  File "/home/am/Projects/venv/cn/bin/nameko", line 11, in <module>
    sys.exit(main())
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/main.py",
line 62, in main
    args.main(args)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/commands.py",
line 85, in main
    main(args)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/run.py",
line 181, in main
    run(services, config, backdoor_port=args.backdoor_port)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/run.py",
line 116, in run
    service_runner.add_service(service_cls)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/runners.py",
line 70, in add_service
    container = self.container_cls(cls, self.config, worker_ctx_cls)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/containers.py",
line 156, in __init__
    bound = dependency.bind(self.interface, attr_name)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/extensions.py",
line 145, in bind
    instance = super(DependencyProvider, self).bind(container)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/extensions.py",
line 100, in bind
    setattr(instance, name, ext.bind(container))
TypeError: bind() missing 1 required positional argument: 'attr_name'

Since I don't find any documentation about publishers and consumer, I
follow the steps I found in here
<Redirecting to Google Groups.
But I still don't know what I'm missing...

Thanks in advance!

Alfonso

···

On Friday, April 7, 2017 at 4:24:44 PM UTC+2, alfo...@gmail.com wrote:

That's it! Thank you Matt!

On Thursday, April 6, 2017 at 1:33:53 PM UTC+2, alfo...@gmail.com wrote:

Hi, I'm trying to use Nameko for the creation of a simple notification
center that sends emails consuming from several queues in RabbitMQ.

Our whole architecture is service based and we have a system to write the
exceptions of every service in our database that are queued in a different
queue in Rabbit too. Is there a way to handle the exceptions that raise
during a worker job? I notice that they are logged and the service is still
running. But what I want is to publish a message in the Error queue when an
exception raises besides the log. I've tried with sys.excepthook and raised
an exception on purpose but the exception was catched before it could be
handled by the excepthook. Any ideas?

Thanks in advance!

Alfonso

5

Hi Matt, I'm having some trouble with this. The idea is that when I catch
the exception with the worker_result, I want to publish a message in the
exchange called ErrorExchange. I have this code:

import traceback

from kombu.messaging import Exchange
from nameko.extensions import DependencyProvider
from nameko.messaging import Publisher

error_exchange = Exchange("ErrorExchange", type='fanout')

class ErrorHandler(DependencyProvider):
    
    p = Publisher(exchange=error_exchange)

    def worker_result(self, worker_ctx, res, exc_info):
        
        if exc_info is None:
            return # nothing to do

        exc_type, exc, tb = exc_info
        msg = { "message": str(exc_info), "stack":
traceback.format_tb(tb), "type": str(exc_type) }
        self.p(msg) #Publish message in the exchange

But when I try to run, I get this error:

(cn) am@VdiUB-02:~/Projects/utils/cn$ nameko run notificationcenter
--broker amqp://guest:guest@localhost
Traceback (most recent call last):
  File "/home/am/Projects/venv/cn/bin/nameko", line 11, in <module>
    sys.exit(main())
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/main.py",
line 62, in main
    args.main(args)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/commands.py",
line 85, in main
    main(args)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/run.py",
line 181, in main
    run(services, config, backdoor_port=args.backdoor_port)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/run.py",
line 116, in run
    service_runner.add_service(service_cls)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/runners.py",
line 70, in add_service
    container = self.container_cls(cls, self.config, worker_ctx_cls)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/containers.py",
line 156, in __init__
    bound = dependency.bind(self.interface, attr_name)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/extensions.py",
line 145, in bind
    instance = super(DependencyProvider, self).bind(container)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/extensions.py",
line 100, in bind
    setattr(instance, name, ext.bind(container))
TypeError: bind() missing 1 required positional argument: 'attr_name'

Since I don't find any documentation about publishers and consumer, I
follow the steps I found in here
<Redirecting to Google Groups.
But I still don't know what I'm missing...

The `Publisher` in your example is a DependencyProvider, and so is the
`ErrorHandler. You can't have DependencyProvider inside a
DependencyProvider, because "dependency injection" only happens for the
service class.

The code you actually need in inside `Publisher.get_dependency`. This is
the bit that would be injected into the service if you added the Publisher
to a service class. You can just lift that out and publish directly inside
the `ErrorHandler`.

I have a branch in progress that pulls all the publishing code out into an
easier-to-reuse package, but for the time being your best bet is to
reimplement/copy it into your `ErrorHandler`.

Thanks in advance!

···

On Monday, April 10, 2017 at 3:39:32 PM UTC+1, alfo...@gmail.com wrote:

Alfonso

On Friday, April 7, 2017 at 4:24:44 PM UTC+2, alfo...@gmail.com wrote:

That's it! Thank you Matt!

On Thursday, April 6, 2017 at 1:33:53 PM UTC+2, alfo...@gmail.com wrote:

Hi, I'm trying to use Nameko for the creation of a simple notification
center that sends emails consuming from several queues in RabbitMQ.

Our whole architecture is service based and we have a system to write
the exceptions of every service in our database that are queued in a
different queue in Rabbit too. Is there a way to handle the exceptions that
raise during a worker job? I notice that they are logged and the service is
still running. But what I want is to publish a message in the Error queue
when an exception raises besides the log. I've tried with sys.excepthook
and raised an exception on purpose but the exception was catched before it
could be handled by the excepthook. Any ideas?

Thanks in advance!

Alfonso

6

Ok! Thanks again for the quick reply! I'll follow your advice and try to
readapt the publisher.get_dependency into my service. I'll be paying
attention at the movements in the Nameko repository. The Nameko framework
is a really good job! Thank you!

Alfonso

···

On Monday, April 10, 2017 at 5:35:18 PM UTC+2, Matt Yule-Bennett wrote:

On Monday, April 10, 2017 at 3:39:32 PM UTC+1, alfo...@gmail.com wrote:

Hi Matt, I'm having some trouble with this. The idea is that when I catch
the exception with the worker_result, I want to publish a message in the
exchange called ErrorExchange. I have this code:

import traceback

from kombu.messaging import Exchange
from nameko.extensions import DependencyProvider
from nameko.messaging import Publisher

error_exchange = Exchange("ErrorExchange", type='fanout')

class ErrorHandler(DependencyProvider):
    
    p = Publisher(exchange=error_exchange)

    def worker_result(self, worker_ctx, res, exc_info):
        
        if exc_info is None:
            return # nothing to do

        exc_type, exc, tb = exc_info
        msg = { "message": str(exc_info), "stack":
traceback.format_tb(tb), "type": str(exc_type) }
        self.p(msg) #Publish message in the exchange

But when I try to run, I get this error:

(cn) am@VdiUB-02:~/Projects/utils/cn$ nameko run notificationcenter
--broker amqp://guest:guest@localhost
Traceback (most recent call last):
  File "/home/am/Projects/venv/cn/bin/nameko", line 11, in <module>
    sys.exit(main())
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/main.py",
line 62, in main
    args.main(args)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/commands.py",
line 85, in main
    main(args)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/run.py",
line 181, in main
    run(services, config, backdoor_port=args.backdoor_port)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/cli/run.py",
line 116, in run
    service_runner.add_service(service_cls)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/runners.py",
line 70, in add_service
    container = self.container_cls(cls, self.config, worker_ctx_cls)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/containers.py",
line 156, in __init__
    bound = dependency.bind(self.interface, attr_name)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/extensions.py",
line 145, in bind
    instance = super(DependencyProvider, self).bind(container)
  File
"/home/am/Projects/venv/cn/lib/python3.5/site-packages/nameko/extensions.py",
line 100, in bind
    setattr(instance, name, ext.bind(container))
TypeError: bind() missing 1 required positional argument: 'attr_name'

Since I don't find any documentation about publishers and consumer, I
follow the steps I found in here
<Redirecting to Google Groups.
But I still don't know what I'm missing...

The `Publisher` in your example is a DependencyProvider, and so is the
`ErrorHandler. You can't have DependencyProvider inside a
DependencyProvider, because "dependency injection" only happens for the
service class.

The code you actually need in inside `Publisher.get_dependency`. This is
the bit that would be injected into the service if you added the Publisher
to a service class. You can just lift that out and publish directly inside
the `ErrorHandler`.

I have a branch in progress that pulls all the publishing code out into an
easier-to-reuse package, but for the time being your best bet is to
reimplement/copy it into your `ErrorHandler`.

Thanks in advance!

Alfonso

On Friday, April 7, 2017 at 4:24:44 PM UTC+2, alfo...@gmail.com wrote:

That's it! Thank you Matt!

On Thursday, April 6, 2017 at 1:33:53 PM UTC+2, alfo...@gmail.com wrote:

Hi, I'm trying to use Nameko for the creation of a simple notification
center that sends emails consuming from several queues in RabbitMQ.

Our whole architecture is service based and we have a system to write
the exceptions of every service in our database that are queued in a
different queue in Rabbit too. Is there a way to handle the exceptions that
raise during a worker job? I notice that they are logged and the service is
still running. But what I want is to publish a message in the Error queue
when an exception raises besides the log. I've tried with sys.excepthook
and raised an exception on purpose but the exception was catched before it
could be handled by the excepthook. Any ideas?

Thanks in advance!

Alfonso

1 Like