diff --git a/swh/journal/client.py b/swh/journal/client.py index 386b400..a55bfc2 100644 --- a/swh/journal/client.py +++ b/swh/journal/client.py @@ -1,133 +1,134 @@ # Copyright (C) 2017 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import logging from abc import ABCMeta, abstractmethod from collections import defaultdict from kafka import KafkaConsumer from swh.core.config import SWHConfig -from .serializers import kafka_to_value +from .serializers import kafka_to_key, kafka_to_value # Only accepted offset reset policy accepted ACCEPTED_OFFSET_RESET = ['earliest', 'latest'] # Only accepted object types ACCEPTED_OBJECT_TYPES = [ 'content', 'revision', 'release', 'occurrence', 'origin', 'origin_visit' ] class SWHJournalClient(SWHConfig, metaclass=ABCMeta): """A base client for the Software Heritage journal. The current implementation of the journal uses Apache Kafka brokers to publish messages under a given topic prefix, with each object type using a specific topic under that prefix. Clients subscribe to events specific to each object type by using the `object_types` configuration variable. Clients can be sharded by setting the `client_id` to a common value across instances. The journal will share the message throughput across the nodes sharing the same client_id. Messages are processed by the `process_objects` method in batches of maximum `max_messages`. """ DEFAULT_CONFIG = { # Broker to connect to 'brokers': ('list[str]', ['localhost']), # Prefix topic to receive notification from 'topic_prefix': ('str', 'swh.journal.objects'), # Consumer identifier 'consumer_identifier': ('str', 'swh.journal.client.test'), # Object types to deal with (in a subscription manner) 'object_types': ('list[str]', [ 'content', 'revision', 'release', 'occurrence', 'origin', 'origin_visit' ]), # Number of messages to batch process 'max_messages': ('int', 100), 'auto_offset_reset': ('str', 'earliest') } CONFIG_BASE_FILENAME = 'journal/client' ADDITIONAL_CONFIG = None def __init__(self, extra_configuration={}): self.config = self.parse_config_file( additional_configs=[self.ADDITIONAL_CONFIG]) if extra_configuration: self.config.update(extra_configuration) self.log = logging.getLogger('swh.journal.client.SWHJournalClient') auto_offset_reset = self.config['auto_offset_reset'] if auto_offset_reset not in ACCEPTED_OFFSET_RESET: raise ValueError( 'Option \'auto_offset_reset\' only accept %s.' % ACCEPTED_OFFSET_RESET) object_types = self.config['object_types'] for object_type in object_types: if object_type not in ACCEPTED_OBJECT_TYPES: raise ValueError( 'Option \'object_types\' only accepts %s.' % ACCEPTED_OFFSET_RESET) self.consumer = KafkaConsumer( bootstrap_servers=self.config['brokers'], + key_deserializer=kafka_to_key, value_deserializer=kafka_to_value, auto_offset_reset=auto_offset_reset, enable_auto_commit=False, group_id=self.config['consumer_identifier'], ) self.consumer.subscribe( topics=['%s.%s' % (self.config['topic_prefix'], object_type) for object_type in object_types], ) self.max_messages = self.config['max_messages'] def process(self): """Main entry point to process event message reception. """ while True: messages = defaultdict(list) for num, message in enumerate(self.consumer): object_type = message.topic.split('.')[-1] messages[object_type].append(message.value) if num >= self.max_messages: break self.process_objects(messages) self.consumer.commit() # Override the following method in the sub-classes @abstractmethod def process_objects(self, messages): """Process the objects (store, compute, etc...) Args: messages (dict): Dict of key object_type (as per configuration) and their associated values. """ pass diff --git a/swh/journal/publisher.py b/swh/journal/publisher.py index 366e633..5364410 100644 --- a/swh/journal/publisher.py +++ b/swh/journal/publisher.py @@ -1,123 +1,125 @@ -# Copyright (C) 2016 The Software Heritage developers +# Copyright (C) 2016-2017 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from collections import defaultdict import logging from kafka import KafkaProducer, KafkaConsumer from swh.core.config import SWHConfig from swh.storage import get_storage -from .serializers import kafka_to_value, value_to_kafka +from .serializers import kafka_to_key, key_to_kafka, value_to_kafka class SWHJournalPublisher(SWHConfig): DEFAULT_CONFIG = { 'brokers': ('list[str]', ['getty.internal.softwareheritage.org']), 'temporary_prefix': ('str', 'swh.tmp_journal.new'), 'final_prefix': ('str', 'swh.journal.test_publisher'), 'consumer_id': ('str', 'swh.journal.publisher.test'), 'publisher_id': ('str', 'swh.journal.publisher.test'), 'object_types': ('list[str]', ['content', 'revision', 'release']), 'storage': ('dict', { 'cls': 'remote', 'args': { 'url': 'http://localhost:5002/', } }), 'max_messages': ('int', 10000), } CONFIG_BASE_FILENAME = 'journal/publisher' def __init__(self, extra_configuration=None): self.config = config = self.parse_config_file() if extra_configuration: config.update(extra_configuration) self.storage = get_storage(**config['storage']) + # yes, the temporary topics contain values that are actually _keys_ self.consumer = KafkaConsumer( bootstrap_servers=config['brokers'], - value_deserializer=kafka_to_value, + value_deserializer=kafka_to_key, auto_offset_reset='earliest', enable_auto_commit=False, group_id=config['consumer_id'], ) self.producer = KafkaProducer( bootstrap_servers=config['brokers'], + key_serializer=key_to_kafka, value_serializer=value_to_kafka, client_id=config['publisher_id'], ) self.consumer.subscribe( topics=['%s.%s' % (config['temporary_prefix'], object_type) for object_type in config['object_types']], ) self.max_messages = self.config['max_messages'] def poll(self): """Process a batch of messages""" num = 0 messages = defaultdict(list) for num, message in enumerate(self.consumer): object_type = message.topic.split('.')[-1] messages[object_type].append(message.value) if num >= self.max_messages: break new_objects = self.process_objects(messages) self.produce_messages(new_objects) self.consumer.commit() def process_objects(self, messages): processors = { 'content': self.process_contents, 'revision': self.process_revisions, 'release': self.process_releases, } return { key: processors[key](value) for key, value in messages.items() } def produce_messages(self, messages): for object_type, objects in messages.items(): topic = '%s.%s' % (self.config['final_prefix'], object_type) for key, object in objects: self.producer.send(topic, key=key, value=object) self.producer.flush() def process_contents(self, content_objs): metadata = self.storage.content_get_metadata(content_objs) return [(content['sha1'], content) for content in metadata] def process_revisions(self, revision_objs): metadata = self.storage.revision_get(revision_objs) return [(revision['id'], revision) for revision in metadata] def process_releases(self, release_objs): metadata = self.storage.release_get(release_objs) return [(release['id'], release) for release in metadata] if __name__ == '__main__': logging.basicConfig( level=logging.INFO, format='%(asctime)s %(process)d %(levelname)s %(message)s' ) publisher = SWHJournalPublisher() while True: publisher.poll() diff --git a/swh/journal/serializers.py b/swh/journal/serializers.py index b8d5c4d..9895757 100644 --- a/swh/journal/serializers.py +++ b/swh/journal/serializers.py @@ -1,16 +1,30 @@ -# Copyright (C) 2016 The Software Heritage developers +# Copyright (C) 2016-2017 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import msgpack +def key_to_kafka(key): + """Serialize a key, possibly a dict, in a predictable way""" + p = msgpack.Packer(use_bin_type=True) + if isinstance(key, dict): + return p.pack_map_pairs(sorted(key.items())) + else: + return p.pack(key) + + +def kafka_to_key(kafka_key): + """Deserialize a key""" + return msgpack.loads(kafka_key) + + def value_to_kafka(value): """Serialize some data for storage in kafka""" return msgpack.dumps(value, use_bin_type=True) def kafka_to_value(kafka_value): """Deserialize some data stored in kafka""" return msgpack.loads(kafka_value) diff --git a/swh/journal/tests/test_serializers.py b/swh/journal/tests/test_serializers.py new file mode 100644 index 0000000..9d4bdd4 --- /dev/null +++ b/swh/journal/tests/test_serializers.py @@ -0,0 +1,29 @@ +# Copyright (C) 2017 The Software Heritage developers +# See the AUTHORS file at the top-level directory of this distribution +# License: GNU General Public License version 3, or any later version +# See top-level LICENSE file for more information + +from collections import OrderedDict +import itertools +import unittest + +from swh.journal import serializers + + +class TestSerializers(unittest.TestCase): + def test_key_to_kafka_repeatable(self): + """Check the kafka key encoding is repeatable""" + base_dict = { + 'a': 'foo', + 'b': 'bar', + 'c': 'baz', + } + + key = serializers.key_to_kafka(base_dict) + + for dict_keys in itertools.permutations(base_dict): + d = OrderedDict() + for k in dict_keys: + d[k] = base_dict[k] + + self.assertEqual(key, serializers.key_to_kafka(d))