-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: pass along event headers as metadata when emitting
- Loading branch information
Rebecca Graber
committed
Jan 31, 2023
1 parent
f997905
commit 6d806b0
Showing
8 changed files
with
199 additions
and
65 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| """ | ||
| Test header conversion utils | ||
| """ | ||
|
|
||
| from datetime import datetime | ||
| from uuid import uuid1 | ||
|
|
||
| import attr | ||
| from django.test import TestCase, override_settings | ||
| from openedx_events.data import EventsMetadata | ||
|
|
||
| from edx_event_bus_kafka.internal.utils import _get_headers_from_metadata, _get_metadata_from_headers | ||
|
|
||
|
|
||
| class TestUtils(TestCase): | ||
| """ Tests for header conversion utils """ | ||
| def test_headers_from_event_metadata(self): | ||
| with override_settings(SERVICE_VARIANT='test'): | ||
| metadata = EventsMetadata(event_type="org.openedx.learning.auth.session.login.completed.v1", | ||
| time=datetime.fromisoformat("2023-01-01T14:00:00+00:00")) | ||
| headers = _get_headers_from_metadata(event_metadata=metadata) | ||
| self.assertDictEqual(headers, { | ||
| 'ce_type': b'org.openedx.learning.auth.session.login.completed.v1', | ||
| 'ce_id': str(metadata.id).encode("utf8"), | ||
| 'ce_source': b'openedx/test/web', | ||
| 'ce_specversion': b'1.0', | ||
| 'sourcehost': metadata.sourcehost.encode("utf8"), | ||
| 'content-type': b'application/avro', | ||
| 'ce_datacontenttype': b'application/avro', | ||
| 'ce_time': b'2023-01-01T14:00:00+00:00', | ||
| 'sourcelib': str(metadata.sourcelib).encode("utf8"), | ||
| 'ce_minorversion': str(metadata.minorversion).encode("utf8") | ||
| }) | ||
|
|
||
| def test_metadata_from_headers(self): | ||
| uuid = uuid1() | ||
| headers = [ | ||
| ('ce_type', b'org.openedx.learning.auth.session.login.completed.v1'), | ||
| ('ce_id', str(uuid).encode("utf8")), | ||
| ('ce_source', b'openedx/test/web'), | ||
| ('ce_specversion', b'1.0'), | ||
| ('sourcehost', b'testsource'), | ||
| ('content-type', b'application/avro'), | ||
| ('ce_datacontenttype', b'application/avro'), | ||
| ('ce_time', b'2023-01-01T14:00:00+00:00'), | ||
| ('sourcelib', b'(1,2,3)') | ||
| ] | ||
| generated_metadata = _get_metadata_from_headers(headers) | ||
| expected_metadata = EventsMetadata( | ||
| event_type="org.openedx.learning.auth.session.login.completed.v1", | ||
| id=uuid, | ||
| minorversion=0, | ||
| source='openedx/test/web', | ||
| sourcehost='testsource', | ||
| time=datetime.fromisoformat("2023-01-01T14:00:00+00:00"), | ||
| sourcelib=(1, 2, 3), | ||
| ) | ||
| self.assertDictEqual(attr.asdict(generated_metadata), attr.asdict(expected_metadata)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| """ | ||
| Utilities for converting between message headers and EventsMetadata | ||
| """ | ||
|
|
||
| import logging | ||
| from collections import defaultdict | ||
| from datetime import datetime | ||
| from typing import List, Tuple | ||
| from uuid import UUID | ||
|
|
||
| from openedx_events.data import EventsMetadata | ||
|
|
||
| EVENT_TYPE_HEADER_KEY = "ce_type" | ||
| ID_HEADER_KEY = "ce_id" | ||
| SOURCE_HEADER_KEY = "ce_source" | ||
| SPEC_VERSION_HEADER_KEY = "ce_specversion" | ||
| TIME_HEADER_KEY = "ce_time" | ||
| MINORVERSION_HEADER_KEY = "ce_minorversion" | ||
|
|
||
| # not CloudEvent headers, so no "ce" prefix | ||
| SOURCEHOST_HEADER_KEY = "sourcehost" | ||
| SOURCELIB_HEADER_KEY = "sourcelib" | ||
|
|
||
| # The documentation is unclear as to which of the following two headers to use for content type, so for now | ||
| # use both | ||
| CONTENT_TYPE_HEADER_KEY = "content-type" | ||
| DATA_CONTENT_TYPE_HEADER_KEY = "ce_datacontenttype" | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| HEADER_KEY_TO_EVENTSMETADATA_FIELD = { | ||
| ID_HEADER_KEY: 'id', | ||
| EVENT_TYPE_HEADER_KEY: 'event_type', | ||
| MINORVERSION_HEADER_KEY: 'minorversion', | ||
| SOURCE_HEADER_KEY: 'source', | ||
| SOURCEHOST_HEADER_KEY: 'sourcehost', | ||
| TIME_HEADER_KEY: 'time', | ||
| SOURCELIB_HEADER_KEY: 'sourcelib' | ||
| } | ||
|
|
||
|
|
||
| def _get_metadata_from_headers(headers: List[Tuple]): | ||
| """ | ||
| Create an EventsMetadata object from the headers of a Kafka message | ||
| Arguments | ||
| headers: The list of headers returned from calling message.headers() on a consumed message | ||
| Returns | ||
| An instance of EventsMetadata with the parameters from the headers. Any fields missing from the headers | ||
| are set to the defaults of the EventsMetadata class | ||
| """ | ||
| # Transform list of (header, value) tuples to a {header: [list of values]} dict. Necessary as an intermediate | ||
| # step because there is no guarantee of unique headers in the list of tuples | ||
| headers_as_dict = defaultdict(list) | ||
| metadata_kwargs = {} | ||
| for key, value in headers: | ||
| headers_as_dict[key].append(value) | ||
|
|
||
| # go through all the headers we care about and set the appropriate field | ||
| for header_key, metadata_field in HEADER_KEY_TO_EVENTSMETADATA_FIELD.items(): | ||
| header_values = headers_as_dict[header_key] | ||
| if len(header_values) == 0: | ||
| logger.warning(f"Missing required \"{header_key}\" header on message, will use EventMetadata default") | ||
| continue | ||
| if len(header_values) > 1: | ||
| logger.warning(f"Multiple \"{header_key}\" headers found on message, using the first one found") | ||
| header_value = header_values[0].decode("utf-8") | ||
| # some headers require conversion to the expected type | ||
| if header_key == ID_HEADER_KEY: | ||
| metadata_kwargs[metadata_field] = UUID(header_value) | ||
| elif header_key == TIME_HEADER_KEY: | ||
| metadata_kwargs[metadata_field] = datetime.fromisoformat(header_value) | ||
| elif header_key == SOURCELIB_HEADER_KEY: | ||
| # convert string "(x,y,z)" to tuple | ||
| metadata_kwargs[metadata_field] = tuple(int(x) for x in header_value[1:-1].split(",")) | ||
| else: | ||
| # these are all string values and don't need any conversion step | ||
| metadata_kwargs[metadata_field] = header_value | ||
| return EventsMetadata(**metadata_kwargs) | ||
|
|
||
|
|
||
| def _get_headers_from_metadata(event_metadata: EventsMetadata): | ||
| """ | ||
| Create a dictionary of CloudEvent-compliant Kafka headers from an EventsMetadata object. | ||
| This method assumes the EventMetadata object was the one sent with the event data to the original signal handler. | ||
| Arguments: | ||
| event_metadata: An EventsMetadata object sent by an OpenEdxPublicSignal | ||
| Returns: | ||
| A dictionary of headers | ||
| """ | ||
| # Dictionary (or list of key/value tuples) where keys are strings and values are binary. | ||
| # CloudEvents specifies using UTF-8; that should be the default, but let's make it explicit. | ||
| return { | ||
| # The way EventMetadata is initialized none of these should ever be null. | ||
| # If it is we want the error to be raised. | ||
| EVENT_TYPE_HEADER_KEY: event_metadata.event_type.encode("utf-8"), | ||
| ID_HEADER_KEY: str(event_metadata.id).encode("utf-8"), | ||
| SOURCE_HEADER_KEY: event_metadata.source.encode("utf-8"), | ||
| SOURCEHOST_HEADER_KEY: event_metadata.sourcehost.encode("utf-8"), | ||
| TIME_HEADER_KEY: event_metadata.time.isoformat().encode("utf-8"), | ||
| MINORVERSION_HEADER_KEY: str(event_metadata.minorversion).encode("utf-8"), | ||
| SOURCELIB_HEADER_KEY: str(event_metadata.sourcelib).encode("utf-8"), | ||
|
|
||
| # Always 1.0. See "Fields" in OEP-41 | ||
| SPEC_VERSION_HEADER_KEY: b'1.0', | ||
| CONTENT_TYPE_HEADER_KEY: b'application/avro', | ||
| DATA_CONTENT_TYPE_HEADER_KEY: b'application/avro', | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters