Page MenuHomeSoftware Heritage
Differential D1992

writer: move the InMemoryJournalWriter from swh.storage in swh.journal
ClosedPublic
Actions

Authored by douardda on Sep 17 2019, 11:40 AM.

Details

Reviewers
vlorentz
ardumont
Group Reviewers
Reviewers
Commits
rDJNL6676f667a5d7: writer: rename writer.direct.DirectKafkaWriter as writer.kafka.
rDJNL9d505445b535: writer: move the InMemoryJournalWriter from swh.storage in swh.journal
Summary

and put both JournalWriter implementations in a swh.journal.writer package.
Also move the get_journal_writer() function from storage.

Keep a swh.journal.direct_writer module for bw compat.

This prevents a dependecy loop between swh.storage and swh.journal.

Diff Detail

Repository
rDJNL Journal infrastructure
Branch
master
Lint
No Linters Available
Unit
No Unit Test Coverage
Build Status
Buildable 7814
Build 11235: tox-on-jenkinsJenkins
Build 11234: arc lint + arc unit

Event Timeline

douardda created this revision.Sep 17 2019, 11:40 AM
Herald added a reviewer: Reviewers. · View Herald TranscriptSep 17 2019, 11:40 AM
douardda added a child revision: D1993: journal_writer: use journal writer from swh.journal.Sep 17 2019, 11:42 AM
swh-public-ci added a comment.Sep 17 2019, 11:43 AM

Build is green
See https://jenkins.softwareheritage.org/job/DJNL/job/tox/251/ for more details.

Harbormaster completed remote builds in B7814: Diff 6714.Sep 17 2019, 11:43 AM
vlorentz added a subscriber: vlorentz.Sep 17 2019, 11:45 AM
vlorentz added inline comments.
swh/journal/writer/__init__.py
11

might as well rename it KafkaWriter while we're at it. "direct" was to be opposed to the pg listener and journal publisher, which were both dropped from the codebase, so it no long makes sense.

vlorentz accepted this revision.Sep 17 2019, 11:47 AM
This revision is now accepted and ready to land.Sep 17 2019, 11:47 AM
ardumont added a subscriber: ardumont.Sep 17 2019, 12:09 PM
ardumont added inline comments.
swh/journal/writer/__init__.py
11

Even JournalWriter directly?
This would be consistent with the in-memory version.

ardumont added inline comments.Sep 17 2019, 12:11 PM
swh/journal/writer/__init__.py
13

I know it's existing stuff we are moving around...
Might as well fix it now though 'Unknown journal writer class'...

ardumont requested changes to this revision.Sep 17 2019, 12:11 PM
This revision now requires changes to proceed.Sep 17 2019, 12:11 PM
douardda updated this revision to Diff 6716.Sep 17 2019, 1:18 PM

writer: rename writer.direct.DirectKafkaWriter as writer.kafka.KafkaJournalWriter

swh-public-ci added a comment.Sep 17 2019, 1:21 PM

Build is green
See https://jenkins.softwareheritage.org/job/DJNL/job/tox/252/ for more details.

Harbormaster completed remote builds in B7816: Diff 6716.Sep 17 2019, 1:21 PM
ardumont accepted this revision.Sep 17 2019, 1:28 PM
This revision is now accepted and ready to land.Sep 17 2019, 1:28 PM
Closed by commit rDJNL9d505445b535: writer: move the InMemoryJournalWriter from swh.storage in swh.journal (authored by douardda). · Explain WhySep 17 2019, 5:33 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathSize
swh/
journal/
91 lines
writer/
15 lines
30 lines
writer/
direct.py
direct_writer.py
2 lines

Diff 6714

View Options

swh/journal/direct_writer.py

  • This file was copied to swh/journal/writer/direct.py.
# Copyright (C) 2019 The Software Heritage developers # Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
import logging # for BW compat
from .writer.direct import DirectKafkaWriter # noqa
from confluent_kafka import Producer
from swh.model.hashutil import DEFAULT_ALGORITHMS
from swh.model.model import BaseModel
from .serializers import key_to_kafka, value_to_kafka
logger = logging.getLogger(__name__)
class DirectKafkaWriter:
"""This class is instantiated and used by swh-storage to write incoming
new objects to Kafka before adding them to the storage backend
(eg. postgresql) itself."""
def __init__(self, brokers, prefix, client_id):
self._prefix = prefix
self.producer = Producer({
'bootstrap.servers': brokers,
'client.id': client_id,
'enable.idempotence': 'true',
})
def send(self, topic, key, value):
self.producer.produce(
topic=topic,
key=key_to_kafka(key),
value=value_to_kafka(value),
)
def flush(self):
self.producer.flush()
# Need to service the callbacks regularly by calling poll
self.producer.poll(0)
def _get_key(self, object_type, object_):
if object_type in ('revision', 'release', 'directory', 'snapshot'):
return object_['id']
elif object_type == 'content':
return object_['sha1'] # TODO: use a dict of hashes
elif object_type == 'skipped_content':
return {
hash: object_[hash]
for hash in DEFAULT_ALGORITHMS
}
elif object_type == 'origin':
return {'url': object_['url'], 'type': object_['type']}
elif object_type == 'origin_visit':
return {
'origin': object_['origin'],
'date': str(object_['date']),
}
else:
raise ValueError('Unknown object type: %s.' % object_type)
def _sanitize_object(self, object_type, object_):
if object_type == 'origin_visit':
return {
**object_,
'date': str(object_['date']),
}
elif object_type == 'origin':
assert 'id' not in object_
return object_
def write_addition(self, object_type, object_, flush=True):
"""Write a single object to the journal"""
if isinstance(object_, BaseModel):
object_ = object_.to_dict()
topic = '%s.%s' % (self._prefix, object_type)
key = self._get_key(object_type, object_)
object_ = self._sanitize_object(object_type, object_)
logger.debug('topic: %s, key: %s, value: %s', topic, key, object_)
self.send(topic, key=key, value=object_)
if flush:
self.flush()
write_update = write_addition
def write_additions(self, object_type, objects, flush=True):
"""Write a set of objects to the journal"""
for object_ in objects:
self.write_addition(object_type, object_, flush=False)
if flush:
self.flush()
View Options

swh/journal/writer/__init__.py

  • This file was added.
# Copyright (C) 2019 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
def get_journal_writer(cls, args={}):
if cls == 'inmemory':
from .inmemory import InMemoryJournalWriter as JournalWriter
elif cls == 'kafka':
from .direct import DirectKafkaWriter as JournalWriter
vlorentzUnsubmitted
Not Done
Inline Actions

might as well rename it KafkaWriter while we're at it. "direct" was to be opposed to the pg listener and journal publisher, which were both dropped from the codebase, so it no long makes sense.

vlorentz: might as well rename it KafkaWriter while we're at it. "direct" was to be opposed to the pg…
ardumontUnsubmitted
Not Done
Inline Actions

Even JournalWriter directly?
This would be consistent with the in-memory version.

ardumont: Even JournalWriter directly? This would be consistent with the in-memory version.
else:
raise ValueError('Unknown storage class `%s`' % cls)
ardumontUnsubmitted
Not Done
Inline Actions

I know it's existing stuff we are moving around...
Might as well fix it now though 'Unknown journal writer class'...

ardumont: I know it's existing stuff we are moving around... Might as well fix it now though 'Unknown…
return JournalWriter(**args)
View Options

swh/journal/writer/direct.py

  • This file was copied from swh/journal/direct_writer.py.
# Copyright (C) 2019 The Software Heritage developers # Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
import logging import logging
from confluent_kafka import Producer from confluent_kafka import Producer
from swh.model.hashutil import DEFAULT_ALGORITHMS from swh.model.hashutil import DEFAULT_ALGORITHMS
from swh.model.model import BaseModel from swh.model.model import BaseModel
from .serializers import key_to_kafka, value_to_kafka from swh.journal.serializers import key_to_kafka, value_to_kafka
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class DirectKafkaWriter: class DirectKafkaWriter:
"""This class is instantiated and used by swh-storage to write incoming """This class is instantiated and used by swh-storage to write incoming
new objects to Kafka before adding them to the storage backend new objects to Kafka before adding them to the storage backend
(eg. postgresql) itself.""" (eg. postgresql) itself."""
▲ Show 20 Lines • Show All 73 Lines • Show Last 20 Lines
View Options

swh/journal/writer/inmemory.py

  • This file was added.
# Copyright (C) 2019 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
import copy
from multiprocessing import Manager
from swh.model.model import BaseModel
logger = logging.getLogger(__name__)
class InMemoryJournalWriter:
def __init__(self):
# Share the list of objects across processes, for RemoteAPI tests.
self.manager = Manager()
self.objects = self.manager.list()
def write_addition(self, object_type, object_):
if isinstance(object_, BaseModel):
object_ = object_.to_dict()
self.objects.append((object_type, copy.deepcopy(object_)))
write_update = write_addition
def write_additions(self, object_type, objects):
for object_ in objects:
self.write_addition(object_type, object_)
About · Developer information · Legal