Page MenuHomeSoftware Heritage
Differential D3441

pytest.ini: Drop filterwarnings which never worked
ClosedPublic
Actions

Authored by ardumont on Jul 7 2020, 12:21 PM.

Details

Reviewers
anlambert
Group Reviewers
Reviewers
Commits
rDSCHce63e6afd01b: pytest.ini: Drop filterwarnings which never worked
Summary

Align pytest.ini back to swh-py-template.

Test Plan

tox

Diff Detail

Repository
rDSCH Scheduling utilities
Branch
master
Lint
No Linters Available
Unit
No Unit Test Coverage
Build Status
Buildable 13406
Build 20502: Phabricator diff pipeline on jenkinsJenkins console · Jenkins
Build 20501: arc lint + arc unit

Event Timeline

ardumont created this revision.Jul 7 2020, 12:21 PM
Herald added a reviewer: Reviewers. · View Herald TranscriptJul 7 2020, 12:21 PM
swh-public-ci added a comment.Jul 7 2020, 12:24 PM

Build is green

Patch application report for D3441 (id=12186)

Rebasing onto b2cbb9ba85...

Current branch diff-target is up to date.
Changes applied before test
commit ce63e6afd01bfc66bdf418602d539369ca4332c8
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Jul 7 12:17:55 2020 +0200

    pytest.ini: Drop filterwarnings which never worked

See https://jenkins.softwareheritage.org/job/DSCH/job/tests-on-diff/40/ for more details.

Harbormaster completed remote builds in B13406: Diff 12186.Jul 7 2020, 12:24 PM
anlambert added a subscriber: anlambert.Jul 7 2020, 1:10 PM
anlambert added inline comments.
pytest.ini
3

It does not work because there is a typo here ;-)

anlambert accepted this revision.Jul 7 2020, 1:13 PM

Looking at the latest CI log of scheduler, the warning to filter does not appear anymore so this can be removed indeed.

This revision is now accepted and ready to land.Jul 7 2020, 1:13 PM
Closed by commit rDSCHce63e6afd01b: pytest.ini: Drop filterwarnings which never worked (authored by ardumont). · Explain WhyJul 7 2020, 1:48 PM
This revision was automatically updated to reflect the committed changes.
ardumont added a commit: rDSCHce63e6afd01b: pytest.ini: Drop filterwarnings which never worked.

Revision Contents
Changeset List

PathSize
2 lines

Diff 12186

View Options

pytest.ini

  • This file was added.
# 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
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 = {
olasdUnsubmitted
Done
Inline Actions

The topic wording is an internal kafka thing that doesn't really matter (except as an implementation detail) for swh.journal consumers.

Proposed wording:

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`.
olasd: The `topic` wording is an internal kafka thing that doesn't really matter (except as an…
# Broker to connect to
'brokers': ('list[str]', ['localhost']),
# Prefix topic to receive notification from
'topic_prefix': ('str', 'swh.journal.test_publisher'),
# Consumer identifier
'consumer_id': ('str', 'swh.journal.client.test'),
olasdUnsubmitted
Not Done
Inline Actions

Should be renamed to match the docstring (or the docstring changed to match)

olasd: Should be renamed to match the docstring (or the docstring changed to match)
# 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']
assert auto_offset_reset in ['earliest', 'latest']
olasdUnsubmitted
Done
Inline Actions

Let's give a proper error message (with a ValueError).

While we're checking the configuration, let's hard-code the list of supported object types and check that the configuration matches (again, raising ValueError on unknown values).

olasd: Let's give a proper error message (with a ValueError). While we're checking the configuration…
self.consumer = KafkaConsumer(
bootstrap_servers=self.config['brokers'],
olasdUnsubmitted
Done
Inline Actions

That variable should be made configurable: we should allow consumer to only care about live events without a (big) catchup phase.

olasd: That variable should be made configurable: we should allow consumer to only care about live…
value_deserializer=kafka_to_value,
auto_offset_reset=auto_offset_reset,
enable_auto_commit=False,
group_id=self.config['consumer_id'],
)
self.consumer.subscribe(
topics=['%s.%s' % (self.config['topic_prefix'], object_type)
for object_type in self.config['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
About · Developer information · Legal