Running services¶

lymph instance --config=$PATH_TO_CONFIG_FILE

$ export LYMPH_NODE_CONFIG="/path/to/lymph/config/.lymph.yml"

interfaces:
    echo:
        class: echo:EchoService

interfaces:
    echo_small_valley:
        class: echo:EchoService
        delay: 1

    echo_large_valley:
        class: echo:EchoService
        delay: 10

$ lymph node

instances:
    echo:
        command: lymph instance --config=conf/echo.yml
        numprocesses: 10

    demo:
        command: lymph instance --config=conf/demo.yml

Configuration¶

Lymph needs to be configured so that it knows how and where to find its service registry and its event system.

container:
    registry:
        class: lymph.discovery.zookeeper:ZookeeperServiceRegistry
        zkclient:
            class: kazoo.client:KazooClient
            hosts: 120.0.0.1:2181

    events:
        class: lymph.events.kombu:KombuEventSystem
        transport: amqp
        hostname: 127.0.0.1

You can find this sample configuration file in conf/sample-node.yml.

key: protocol://$(env.USER):$(env.PASSWORD)@host/path

# vars.yml
key: value
struct:
    foo: bar

# conf.yml
foo: $(var.key)
var: $(var.struct)
interpolation: prefix_$(var.key)_suffix

dependencies:
    <name>:
        class: <class path>
        <extra class arguments>

container:
    metrics:
        tags:
            env: $(env.NAMESPACE)

components:
    SerialEventHandler:
        zkclient: dep:kazoo

Tests¶

You can test if your installation of lymph has been successful by running the unittests. You’ll also have to set ZOOKEEPER_PATH to the directory that contains your ZooKeeper binaries (e.g. /usr/share/java on Ubuntu).

You can then run the tests with either `tox`_ or nosetests directly.

Events¶

container:
    events:
        class: lymph.events.kombu:KombuEventSystem
        transport: amqp
        hostname: 127.0.0.1

class SomeClass(lymph.Interface):
        def emit_event(self):
                self.emit('simple_event', {'article': 'foo', 'quantity': 5})

$ lymph subscribe

$ lymph emit

$ lymph subscribe uppercase_transform_finished
uppercase_transform_finished: {'text': u'foo_282'}
uppercase_transform_finished: {'text': u'foo_283'}
uppercase_transform_finished: {'text': u'foo_284'}
…

$ lymph subscribe event_a event_b
event_a: {u'data': u'nice'}
event_b: {u'information': u'data'}

$ lymph emit uppercase_transform_finished '{"text": "bar_foo_234"}'

$ lymph subscribe uppercase_transform_finished
…
uppercase_transform_finished: {'text': u'foo_2629'}
uppercase_transform_finished: {'text': u'foo_2630'}
uppercase_transform_finished: {u'text': u'bar_foo_234'}
uppercase_transform_finished: {'text': u'foo_2631'}
…

RPC¶

import lymph

class EchoService(lymph.Interface):
    pass

interfaces:
    echo:
        class: project.interfaces:EchoService

def some_rpc_method(self, channel, **kwargs):
    …

HTTP¶

from lymph.web.interfaces import WebServiceInterface
from werkzeug.routing import Map, Rule
from werkzeug.wrappers import Response


class HttpHello(WebServiceInterface)
    url_map = Map([
        Rule('/hello/<string:name>/', endpoint='hello'),
    ])

    def hello(self, request, name):
        return Response('hello %s!' % name)

class WebServiceInterface¶

is_healthy()¶

Serialization¶

class Process(object):
    ...

    def _lymph_dump_(self):
        return {
            'pid': self.pid,
            'name': self.name,
        }


class Node(lymph.Interface):

    @lymph.rpc()
    def get_processes(self, service_type=None):
        procs = []
        for proc in self._processes:
            if not service_type or proc.service_type == service_type:
                procs.append(proc)
        return procs

    @lymph.rpc()
    def stop(self, service_type=None):
        for proc in self.get_processes(service_type):
            proc.stop()

Versioning interfaces¶

interfaces:
    echo@1.5.0:
        class: echo:Echo

    echo@2.0.0:
        class: echo:Echo2

$ lymph request echo.upper@1.2 '{"text": "foo"}'

proxy = lymph.proxy('echo', version='1.1')

Service API¶

import lymph


class Echo(lymph.Interface):

    @lymph.rpc()
    def echo(self, text=None):
        return text

    @lymph.rpc()
    def upper(self, text=None):
        self.emit('uppercase_transform_finished', {'text': text})
        return text.upper()

    @lymph.event('uppercase_transform_finished')
    def on_uppercase(self, text=None):
        print "done", text

class lymph.Interface¶

name¶

The interface identifier that is used to register this service with the coordinator service. name is an instance attribute which is taken from the config of the interface taken from the config of the interface.

on_start()¶

Called when the service is started

on_stop()¶

Called when the service is stopped

apply_config(config)¶

Parameters:	config – dict

Called with instance specific configuration that is usually provided by a config file (see Metrics Configuration).

request(address, method, body)¶

Parameters:	address – the address where the request is sent to; either a ZeroMQ endpoint or a service name method – the remote method that will be called body – JSON serializable dict of parameters for the remote method

proxy(address)¶

returns a proxy object that can be used to conveniently send requests to another service.

echo = self.proxy('echo')
result = echo.upper(text='foo')
assert result == 'FOO'

This is equivalent to self.request('echo', 'echo.upper', text='foo').

emit(event_type, payload, delay=0)¶

Parameters:	event_type – str payload – a dict of JSON serializable data structures delay – delay delivery of this event by delay seconds

@subscribe(*event_types, sequential=True)¶

Behaves like lymph.event(), but can be used at runtime

class Example(lymph.Service):
    def on_start(self):
        @self.subscribe('dynamic_event_type')
        def on_event(event):
            assert isinstance(event, lymph.core.events.Event)

@lymph.raw_rpc¶

Marks the decorated interface method as an RPC method. Using this decorator the RPC function are expected to accept a ReplyChannel instance as a first argument.

import lymph

class Example(lymph.Interface):
    @lymph.raw_rpc()
    def do_something(self, channel, message):
        assert isinstance(channel, lymph.core.channels.ReplyChannel)
        assert isinstance(message, lymph.core.messages.Message)
        channel.ack()

@lymph.rpc¶

Marks the decorated interface method as an RPC method. The difference between this decorator and raw_rpc() is that the RPC functions must use return and raise like any normal Python function instead of using channel.reply and channel.error.

Parameters:	raises – tuple of exception classes that the RPC function is expected to raise.

import lymph

class Example(lymph.Interface):
    @lymph.rpc()
    def do_something(self, message):
        return message

@lymph.event(*event_types, sequential=False)¶

Parameters:	event_types – may contain wildcards, e.g. 'subject.*' sequential – force sequential event consumption broadcast – receive every event in all instances

Marks the decorated interface method as an event handler. The service container will automatically subscribe to given event_types. If sequential=True, events will be not be consumed in parallel, but one by one. If broadcast=True, every instance of the service will receive the event.

import lymph

class Example(lymph.Interface):
    @lymph.event('task_done')
    def on_task_done(self, event):
        assert isinstance(event, lymph.core.events.Event)

@lymph.task¶

Parameters:	sequential – force sequential task execution per instance broadcast – execute the task in all instances

Marks the decorated interface method as a task handler.

Core API¶

class lymph.core.container.ServiceContainer¶

classmethod from_config(config, **kwargs)¶

start()¶

stop()¶

send_message(address, msg)¶

Parameters:	address – the address for this message; either a ZeroMQ endpoint a service name msg – the lymph.core.messages.Message object that will be sent
Returns:	lymph.core.channels.ReplyChannel

lookup(address)¶

Parameters:	address – an lymph address
Returns:	lymph.core.services.Service or lymph.core.services.ServiceInstance

class lymph.core.channels.ReplyChannel¶

reply(body)¶

Parameters:	body – a JSON serializable data structure

ack()¶

acknowledges the request message

class lymph.core.channels.RequestChannel¶

get(timeout=1)¶

Returns:	lymph.core.messages.Message

returns the next reply message from this channel. Blocks until the reply is available. Raises Timeout after timeout seconds.

class lymph.core.messages.Message¶

id¶

type¶

subject¶

body¶

packed_body¶

class lymph.core.events.Event¶

type¶

the event type / name

body¶

dictionary with the payload of the message

source¶

id of the event source service

__getitem__(name)¶

gets an event parameter from the body

class lymph.core.services.Service¶

Normally created by ServiceContainer.lookup(). Service objects represent lymph services.

__iter__()¶

Yields all known instances of this service.

__len__()¶

Returns the number of known instances of this service.

class lymph.core.services.ServiceInstance¶

Describes a single service instance. Normally created by ServiceContainer.lookup()

identity¶

The identity string of this service instance

endpoint¶

The rpc endpoint for this

class lymph.core.connections.Connection¶

You can attain a connection to an lymph service instance directly from lymph.core.container.ServiceContainer.connect(), or from the higher-level API in lymph.core.services. For ZeroMQ endpoint addresses the following to statements are roughly equivalent:

container.connect(address)  # only works for tcp://… addresses
container.lookup(address).connect()  # will also work for service names

class lymph.core.interfaces.Proxy(container, address, namespace=None, timeout=1)¶

__getattr__(self, name)¶

Returns a callable that will execute the RPC method with the given name.

Config API¶

class lymph.config.ConfigView(config, prefix)¶

A ConfigView allows access to a subtree of a Configuration object. It implements the mapping protocol. Dotted path keys are translated into nested dictionary lookups, i.e. cv.get('a.b') is (roughly) equivalent to cv.get('a').get('b').

If a value returned by ConfigView methods is a dict, it will be wrapped in a ConfigView itself. This – and getting dicts from a Configuration object – are the preferred way to create new ConfigViews.

root¶

A reference to the root Configuration instance.

class lymph.config.Configuration(values=None)¶

Parameters:	values – an optional initial mapping

Configuration implements the same interface as ConfigView in addition to the methods described here.

load(file, sections=None)¶

Reads yaml configuration from a file-like object. If sections is not None, only the keys given are imported

load_file(path, sections=None)¶

Reads yaml configuration from the file at path.

get_raw(key, default)¶

Like get(), but doesn’t wrap dict values in ConfigView.

create_instance(key, default_class=None, **kwargs)¶

Parameters:	key – dotted config path (e.g. "container.rpc") default_class – class object or fully qualified name of a class kwargs – extra keyword arguments to be passed to the factory

Creates an object from the config dict at key. The instance is created by a factory that is specified by its fully qualified name in a class key of the config dict.

If the factory has a from_config() method it is called with a ConfigView of key. Otherwise, the factory is called directly with the config values as keyword arguments.

Extra keyword arguments to create_instance() are passed through to from_config() or mixed into the arguments if the factory is a plain callable.

If the config doesn’t have a class key the instance is create by default_class, which can be either a fully qualifed name or a factory object.

Given the following config file

foo:
    class: pack.age:SomeClass
    extra_arg: 42

you can create an instance of SomeClass

# in pack/age.py
class SomeClass(object):
    @classmethod
    def from_config(cls, config, **kwargs):
        assert config['extra_arg'] == 42
        assert kwargs['bar'] is True
        return cls(...)

# in any module
config = Configuration()
config.load(...)
config.create_instance('foo', bar=True)

get_instance(key, default_class, **kwargs)¶

Like create_instance(), but only creates a single instance for each key.

Web API¶

class lymph.web.WebServiceInterface¶

application¶

WSGI application instance that this interface is running

url_map¶

A werkzeug.routing.Map instance that is used to map requests to request handlers. Typically given as a class attribute.

Pattern API¶

@lymph.patterns.serial_events.serial_event(*event_types, partition_count=12, key=None)¶

Parameters:	event_types – event types that should be partitioned partition_count – number of queues that should be used to partition the events key – a function that maps Events to string keys. This function should have two arguments in its signature: the instance of current Interface and instance of the handled Event object.

This event handler redistributes events into partition_count queues. These queues are then partitioned over all service instances and consumed sequentially, i.e. at most one event per queue at a time.

Components API¶

Components are objects that depend on a running service container. They are embedded in Componentized objects. Since Componentized objects themselves are components, they form a tree of Component instances with the container as the root. An example of a Component is lymph.core.interfaces.Interface.

class lymph.core.components.Component(error_hook=None, pool=None, metrics=None)¶

error_hook¶

A Hook object that propagates exceptions for this component. Defaults to the error_hook of the parent component.

pool¶

A pool that holds greenlets related to the component. Defaults to the pool of the parent component.

metrics¶

An Aggregate of metrics for this component. Defaults to the metrics of the parent component.

on_start()¶

Called when the container is started.

on_stop()¶

Called when the container is stopped.

spawn(func, *args, **kwargs)¶

Spawns a new greenlet in the greenlet pool of this component. If func exits with an exception, it is reported to the error_hook.

class lymph.core.components.Componentized¶

A collection of components; itself a component.

add_component(component)¶

Parameters:	component – Component

Adds component.

on_start()¶

Calls on_start() on all added components.

on_stop()¶

Calls on_stop() on all added components.

Metrics API¶

To follow the metrics protocol objects must be iterable repeatedly and yield (name, value, tags)-triples, where name is a string, value is a float or int, and tags is a dict with string keys and values.

class lymph.core.monitoring.metrics.Metric(name, tags=None)¶

An abstract base class for single series metrics, i.e. metric objects that only yield a single triple.

__iter__()¶

[abstract] Yields metric values as a tuple in the form (name, value, tags).

class lymph.core.monitoring.metrics.Gauge(name, value=0, tags=None)¶

A gauge is a metric that represents a single numerical value that can arbitrarily go up and down.

set(value)¶

class lymph.core.monitoring.metrics.Callable(name, func, tags=None)¶

Like a Gauge metric, but its value is determined by a callable.

class lymph.core.monitoring.metrics.Counter(name, tags=None)¶

A counter is a cumulative metric that represents a single numerical value that only ever goes up. A counter is typically used to count requests served, tasks completed, errors occurred, etc.

__iadd__(value)¶

Increment counter value.

class lymph.core.monitoring.metrics.TaggedCounter(name, tags=None)¶

A tagged counter is a container metric that represents multiple counters per tags. A tagged counter is typically used to track a group of counters as one e.g. request served per function name, errors ocurred per exception name, etc.

incr(_by=1, **tags)¶

Increment given counter type by _by.

class lymph.core.monitoring.metrics.Aggregate(metrics=(), tags=None)¶

Parameters:	metrics – iterable of metric objects tags – dict of tags to add to all metrics.

Aggregates a collection of metrics into a single metrics object.

add(metric)¶

Parameters:	metric – metric object

Adds the given metric to collection.

add_tags(**tags)¶

Parameters:	tags – string-valued dict

Adds the given tags for all metrics.

Testings API¶

class lymph.testing.RpcMockTestCase¶

Base mixin test class that provice a highlevel interface for mocking remote rpc calls. By inheriting this class, test cases can supply mock return values for rpc functions.

Note

In case an rpc function is not mocked the actual RPC call will be made to the service.

rpc_mock_calls¶

A List of the called rpc functions.

setup_rpc_mocks(rpc_functions)¶

Setup RPC mocks by passing all mocked RPC functions as a dictionary in the form {'<service_name>.<function_name>': <return_value>}, in case <return_value> is an exception, call will raise the exception.

class SomeTest(RpcMockTestCase):

    def setUp(self):
        super().setUp()
        self.setup_rpc_mocks({
            'upper.upper': 'HELLO WORLD',
            'upper.echo': TypeError('...')
            ...
        })

update_rpc_mock(func_name, return_value)¶

Update a mock of an already mocked RPC function.

class SomeTest(RpcMockTestCase):

    def setUp(self):
        super().setUp()
        self.setup_rpc_mocks({
            'upper.upper': 'HELLO WORLD',
            'upper.echo': 'hello world',
        })

   def test_something(self):
        self.update_rpc_mock('upper.upper', 'A NEW VALUE')
        ...

delete_rpc_mock(func_name)¶

Delete a mock of an already mocked RPC function.

Raises:	KeyError – In case the functions wasn’t mocked previously.

class SomeTest(RpcMockTestCase):

    def setUp(self):
        super().setUp()
        self.setup_rpc_mocks({
            'upper.upper': 'HELLO WORLD',
            'upper.echo': 'hello world',
        })

   def test_really_something(self):
        self.delete_rpc_mock('upper.upper')
        ...

assert_rpc_calls(*expected_calls)¶

This method is a convenient way of asserting that rpc function calls were made in a particular way:

class SomeTest(RpcMockTestCase):

    def setUp(self):
        super().setUp()
        self.setup_rpc_mocks({
            'upper.upper': 'HELLO WORLD',
            'upper.echo': 'hello world',
        })

   def test_something(self):
       ...

       self.assert_rpc_calls(
           mock.call('upper.upper', text='hello world')
       )

mock.call(..) can contain PyHamcrest matchers for better and less brittle tests.

assert_any_rpc_calls(*expected_calls)¶

At the opposite of assert_rpc_calls where you have to specify all mocked calls that were done, this method accept a list of mocked calls and assert that each one of them was done. The calls should be specified in the same order as they are made.

Note

This method do it’s best to guess which function user is looking for and in case of a mismatch it try to generate a useful message for the user.

class lymph.testing.EventMockTestCase¶

Base mixin test class that provides a highlevel interface for mocking events emitted.

events¶

A List of the emitted events.

assert_events_emitted(*expected_emitted)¶

This method is a convenient way of asserting that events were emitted:

class SomeTest(EventMockTestCase):

   def test_something(self):
       ...

       self.assert_events_emitted(
           mock.call('upper.uppercase_transform_finished', {'text': 'hello world'})
       )

class lymph.testing.RPCServiceTestCase¶

Test class for testing a unique RPC interface.

service_class¶

Interface class to test, this attribute is abstract and must be supplied by child class.

service_config¶

Configuration to pass to service when calling apply_config().

client¶

Shortcut for getting default lymph.Proxy instance for the service under test.

get_proxy(**kwargs)¶

Return a lymph.Proxy instance of the service under test.

request(*args, **kwargs)¶

Low level method to send a request to service under tests return ReplyChannel instance.

emit(*args, **kwargs)¶

Emit an event.

class lymph.testing.WebServiceTestCase¶

Test class for testing a unique Web interface.

client¶

Return a Werkzeug test client associated to web interface under test.