Page MenuHomeSoftware Heritage
Differential D914

Notify Kafka when an origin visit is update.
ClosedPublic
Actions

Authored by vlorentz on Jan 10 2019, 4:26 PM.

Details

Reviewers
olasd
Group Reviewers
Reviewers
Commits
rDSTOCd19392c0fadb: Notify Kafka when an origin visit is update.
R65:d19392c0fadb: Notify Kafka when an origin visit is update.
rDSTOd19392c0fadb: Notify Kafka when an origin visit is update.
Summary

First step for T1465.

Note that it changes the value of topic_prefix.

Test Plan

There is a new test

Diff Detail

Repository
rDSTO Storage manager
Branch
notify-origin-visit-update
Lint
No Linters Available
Unit
No Unit Test Coverage
Build Status
Buildable 3394
Build 4377: tox-on-jenkinsJenkins
Build 4376: arc lint + arc unit

Event Timeline

vlorentz created this revision.Jan 10 2019, 4:26 PM
Herald added a reviewer: Reviewers. · View Herald TranscriptJan 10 2019, 4:26 PM
vlorentz edited the summary of this revision. (Show Details)Jan 10 2019, 4:26 PM
vlorentz edited the test plan for this revision. (Show Details)
swh-public-ci added a subscriber: swh-public-ci.Jan 10 2019, 4:29 PM

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

Harbormaster completed remote builds in B3388: Diff 2908.Jan 10 2019, 4:29 PM
vlorentz mentioned this in D916: Add support for object updates to the journal publisher..Jan 10 2019, 4:34 PM
olasd added a subscriber: olasd.Jan 10 2019, 4:41 PM

I don't think there's any need to create the changed topics, or to add the mutable objects config : while its name doesn't reflect that, the new topic has always been designed as a way to notify the producer component of objects that need to be updated in the topic the actual clients subscribe to.

That's also apparent in D916 where the changed topic just ends up being used for feeding the same "final" topic.

olasd added a comment.Jan 10 2019, 4:46 PM

To drive the point further : the journal hasn't been designed as a feed of all changes to the archive; it's currently designed as a way to get *the latest version* of all the objects that have been added to the archive, in a somewhat streaming fashion.

In the topics that the consumers are listening to, Kafka compaction rules will make messages about "old" versions of objects disappear when a new version (with an identical key) appears.

vlorentz updated this revision to Diff 2912.Jan 10 2019, 4:48 PM
  • Drop the 'changed' topic.
olasd accepted this revision.Jan 10 2019, 4:50 PM
olasd added inline comments.
swh/storage/listener.py
69

maybe "Registered to events for object type %s" ?

This revision is now accepted and ready to land.Jan 10 2019, 4:50 PM
vlorentz updated this revision to Diff 2913.Jan 10 2019, 4:51 PM
  • Improve log message
vlorentz updated this revision to Diff 2914.Jan 10 2019, 4:51 PM
  • rebase + squash
Closed by commit rDSTOd19392c0fadb: Notify Kafka when an origin visit is update. (authored by vlorentz). · Explain WhyJan 10 2019, 4:51 PM
This revision was automatically updated to reflect the committed changes.
swh-public-ci added a comment.Jan 10 2019, 4:52 PM

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

Harbormaster completed remote builds in B3392: Diff 2912.Jan 10 2019, 4:52 PM
Harbormaster failed remote builds in B3393: Diff 2913!
swh-public-ci added a comment.Jan 10 2019, 4:52 PM

Build has FAILED

Link to build: https://jenkins.softwareheritage.org/job/DSTO/job/tox/154/
See console output for more information: https://jenkins.softwareheritage.org/job/DSTO/job/tox/154/console

Harbormaster failed remote builds in B3394: Diff 2914!Jan 10 2019, 4:52 PM
swh-public-ci added a comment.Jan 10 2019, 4:52 PM

Build has FAILED

Link to build: https://jenkins.softwareheritage.org/job/DSTO/job/tox/155/
See console output for more information: https://jenkins.softwareheritage.org/job/DSTO/job/tox/155/console

Revision Contents
Changeset List

PathSize
swh/
storage/
15 lines
sql/
20 lines
tests/
62 lines

Diff 2914

View Options

swh/storage/listener.py

# Copyright (C) 2016-2017 The Software Heritage developers # Copyright (C) 2016-2017 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 datetime
from enum import Enum
from typing import Any, Union from typing import Any, Union
import msgpack import msgpack
from swh.core.api.serializers import msgpack_dumps, msgpack_loads
from swh.model.model import KeyType from swh.model.model import KeyType
class MsgpackExtTypeCodes(Enum):
LONG_INT = 1
LONG_NEG_INT = 2
# this as been copied from swh.core.api.serializer
# TODO refactor swh.core to make this function available
def _msgpack_encode_int(value):
# needed because msgpack will not handle long integers with more than 64 bits
# which we unfortunately happen to have to deal with from time to time
if value > 0:
code = MsgpackExtTypeCodes.LONG_INT.value
else:
code = MsgpackExtTypeCodes.LONG_NEG_INT.value
value = -value
length, rem = divmod(value.bit_length(), 8)
if rem:
length += 1
return msgpack.ExtType(code, int.to_bytes(value, length, "big"))
def msgpack_ext_encode_types(obj):
if isinstance(obj, int):
return _msgpack_encode_int(obj)
return obj
def msgpack_ext_hook(code, data):
if code == MsgpackExtTypeCodes.LONG_INT.value:
return int.from_bytes(data, "big")
if code == MsgpackExtTypeCodes.LONG_NEG_INT.value:
return -int.from_bytes(data, "big")
raise ValueError("Unknown msgpack extended code %s" % code)
# for BW compat
def decode_types_bw(obj):
if set(obj.keys()) == {b"d", b"swhtype"} and obj[b"swhtype"] == "datetime":
return datetime.datetime.fromisoformat(obj[b"d"])
return obj
def stringify_key_item(k: str, v: Union[str, bytes]) -> str: def stringify_key_item(k: str, v: Union[str, bytes]) -> str:
"""Turn the item of a dict key into a string""" """Turn the item of a dict key into a string"""
if isinstance(v, str): if isinstance(v, str):
return v return v
if k == "url": if k == "url":
return v.decode("utf-8") return v.decode("utf-8")
return v.hex() return v.hex()
Show All 22 Lines
def kafka_to_key(kafka_key: bytes) -> KeyType: def kafka_to_key(kafka_key: bytes) -> KeyType:
"""Deserialize a key""" """Deserialize a key"""
return msgpack.loads(kafka_key, raw=False) return msgpack.loads(kafka_key, raw=False)
def value_to_kafka(value: Any) -> bytes: def value_to_kafka(value: Any) -> bytes:
"""Serialize some data for storage in kafka""" """Serialize some data for storage in kafka"""
return msgpack_dumps(value) return msgpack.packb(
value,
use_bin_type=True,
datetime=True, # encode datetime as msgpack.Timestamp
default=msgpack_ext_encode_types,
)
def kafka_to_value(kafka_value: bytes) -> Any: def kafka_to_value(kafka_value: bytes) -> Any:
"""Deserialize some data stored in kafka""" """Deserialize some data stored in kafka"""
value = msgpack_loads(kafka_value) value = msgpack.unpackb(
kafka_value,
raw=False,
object_hook=decode_types_bw,
ext_hook=msgpack_ext_hook,
strict_map_key=False,
timestamp=3, # convert Timestamp in datetime objects (tz UTC)
)
return ensure_tuples(value) return ensure_tuples(value)
def ensure_tuples(value: Any) -> Any: def ensure_tuples(value: Any) -> Any:
if isinstance(value, (tuple, list)): if isinstance(value, (tuple, list)):
return tuple(map(ensure_tuples, value)) return tuple(map(ensure_tuples, value))
elif isinstance(value, dict): elif isinstance(value, dict):
return dict(ensure_tuples(list(value.items()))) return dict(ensure_tuples(list(value.items())))
else: else:
return value return value
View Options

swh/storage/sql/70-swh-triggers.sql

# Copyright (C) 2016-2017 The Software Heritage developers # Copyright (C) 2016-2017 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 datetime
from enum import Enum
from typing import Any, Union from typing import Any, Union
import msgpack import msgpack
from swh.core.api.serializers import msgpack_dumps, msgpack_loads
from swh.model.model import KeyType from swh.model.model import KeyType
class MsgpackExtTypeCodes(Enum):
LONG_INT = 1
LONG_NEG_INT = 2
# this as been copied from swh.core.api.serializer
# TODO refactor swh.core to make this function available
def _msgpack_encode_int(value):
# needed because msgpack will not handle long integers with more than 64 bits
# which we unfortunately happen to have to deal with from time to time
if value > 0:
code = MsgpackExtTypeCodes.LONG_INT.value
else:
code = MsgpackExtTypeCodes.LONG_NEG_INT.value
value = -value
length, rem = divmod(value.bit_length(), 8)
if rem:
length += 1
return msgpack.ExtType(code, int.to_bytes(value, length, "big"))
def msgpack_ext_encode_types(obj):
if isinstance(obj, int):
return _msgpack_encode_int(obj)
return obj
def msgpack_ext_hook(code, data):
if code == MsgpackExtTypeCodes.LONG_INT.value:
return int.from_bytes(data, "big")
if code == MsgpackExtTypeCodes.LONG_NEG_INT.value:
return -int.from_bytes(data, "big")
raise ValueError("Unknown msgpack extended code %s" % code)
# for BW compat
def decode_types_bw(obj):
if set(obj.keys()) == {b"d", b"swhtype"} and obj[b"swhtype"] == "datetime":
return datetime.datetime.fromisoformat(obj[b"d"])
return obj
def stringify_key_item(k: str, v: Union[str, bytes]) -> str: def stringify_key_item(k: str, v: Union[str, bytes]) -> str:
"""Turn the item of a dict key into a string""" """Turn the item of a dict key into a string"""
if isinstance(v, str): if isinstance(v, str):
return v return v
if k == "url": if k == "url":
return v.decode("utf-8") return v.decode("utf-8")
return v.hex() return v.hex()
Show All 22 Lines
def kafka_to_key(kafka_key: bytes) -> KeyType: def kafka_to_key(kafka_key: bytes) -> KeyType:
"""Deserialize a key""" """Deserialize a key"""
return msgpack.loads(kafka_key, raw=False) return msgpack.loads(kafka_key, raw=False)
def value_to_kafka(value: Any) -> bytes: def value_to_kafka(value: Any) -> bytes:
"""Serialize some data for storage in kafka""" """Serialize some data for storage in kafka"""
return msgpack_dumps(value) return msgpack.packb(
value,
use_bin_type=True,
datetime=True, # encode datetime as msgpack.Timestamp
default=msgpack_ext_encode_types,
)
def kafka_to_value(kafka_value: bytes) -> Any: def kafka_to_value(kafka_value: bytes) -> Any:
"""Deserialize some data stored in kafka""" """Deserialize some data stored in kafka"""
value = msgpack_loads(kafka_value) value = msgpack.unpackb(
kafka_value,
raw=False,
object_hook=decode_types_bw,
ext_hook=msgpack_ext_hook,
strict_map_key=False,
timestamp=3, # convert Timestamp in datetime objects (tz UTC)
)
return ensure_tuples(value) return ensure_tuples(value)
def ensure_tuples(value: Any) -> Any: def ensure_tuples(value: Any) -> Any:
if isinstance(value, (tuple, list)): if isinstance(value, (tuple, list)):
return tuple(map(ensure_tuples, value)) return tuple(map(ensure_tuples, value))
elif isinstance(value, dict): elif isinstance(value, dict):
return dict(ensure_tuples(list(value.items()))) return dict(ensure_tuples(list(value.items())))
else: else:
return value return value
View Options

swh/storage/tests/test_listener.py

# Copyright (C) 2016-2017 The Software Heritage developers # Copyright (C) 2016-2017 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 datetime
from enum import Enum
from typing import Any, Union from typing import Any, Union
import msgpack import msgpack
from swh.core.api.serializers import msgpack_dumps, msgpack_loads
from swh.model.model import KeyType from swh.model.model import KeyType
class MsgpackExtTypeCodes(Enum):
LONG_INT = 1
LONG_NEG_INT = 2
# this as been copied from swh.core.api.serializer
# TODO refactor swh.core to make this function available
def _msgpack_encode_int(value):
# needed because msgpack will not handle long integers with more than 64 bits
# which we unfortunately happen to have to deal with from time to time
if value > 0:
code = MsgpackExtTypeCodes.LONG_INT.value
else:
code = MsgpackExtTypeCodes.LONG_NEG_INT.value
value = -value
length, rem = divmod(value.bit_length(), 8)
if rem:
length += 1
return msgpack.ExtType(code, int.to_bytes(value, length, "big"))
def msgpack_ext_encode_types(obj):
if isinstance(obj, int):
return _msgpack_encode_int(obj)
return obj
def msgpack_ext_hook(code, data):
if code == MsgpackExtTypeCodes.LONG_INT.value:
return int.from_bytes(data, "big")
if code == MsgpackExtTypeCodes.LONG_NEG_INT.value:
return -int.from_bytes(data, "big")
raise ValueError("Unknown msgpack extended code %s" % code)
# for BW compat
def decode_types_bw(obj):
if set(obj.keys()) == {b"d", b"swhtype"} and obj[b"swhtype"] == "datetime":
return datetime.datetime.fromisoformat(obj[b"d"])
return obj
def stringify_key_item(k: str, v: Union[str, bytes]) -> str: def stringify_key_item(k: str, v: Union[str, bytes]) -> str:
"""Turn the item of a dict key into a string""" """Turn the item of a dict key into a string"""
if isinstance(v, str): if isinstance(v, str):
return v return v
if k == "url": if k == "url":
return v.decode("utf-8") return v.decode("utf-8")
return v.hex() return v.hex()
Show All 22 Lines
def kafka_to_key(kafka_key: bytes) -> KeyType: def kafka_to_key(kafka_key: bytes) -> KeyType:
"""Deserialize a key""" """Deserialize a key"""
return msgpack.loads(kafka_key, raw=False) return msgpack.loads(kafka_key, raw=False)
def value_to_kafka(value: Any) -> bytes: def value_to_kafka(value: Any) -> bytes:
"""Serialize some data for storage in kafka""" """Serialize some data for storage in kafka"""
return msgpack_dumps(value) return msgpack.packb(
value,
use_bin_type=True,
datetime=True, # encode datetime as msgpack.Timestamp
default=msgpack_ext_encode_types,
)
def kafka_to_value(kafka_value: bytes) -> Any: def kafka_to_value(kafka_value: bytes) -> Any:
"""Deserialize some data stored in kafka""" """Deserialize some data stored in kafka"""
value = msgpack_loads(kafka_value) value = msgpack.unpackb(
kafka_value,
raw=False,
object_hook=decode_types_bw,
ext_hook=msgpack_ext_hook,
strict_map_key=False,
timestamp=3, # convert Timestamp in datetime objects (tz UTC)
)
return ensure_tuples(value) return ensure_tuples(value)
def ensure_tuples(value: Any) -> Any: def ensure_tuples(value: Any) -> Any:
if isinstance(value, (tuple, list)): if isinstance(value, (tuple, list)):
return tuple(map(ensure_tuples, value)) return tuple(map(ensure_tuples, value))
elif isinstance(value, dict): elif isinstance(value, dict):
return dict(ensure_tuples(list(value.items()))) return dict(ensure_tuples(list(value.items())))
else: else:
return value return value
About · Developer information · Legal