QChris has submitted this change and it was merged.
Change subject: Correcting errors with batching code
......................................................................
Correcting errors with batching code
Removing extra run from PeriodicThread that was
causing recursion exception.
Making sure the main process exits if
an uncaught exception is thrown in the threads spwaned.
Grouping together when batching events that belong to the same
schema and that have identical fields.
Tested in beta-labs.
Bug: 67450
Change-Id: I0f160ef25d05f6b218b97cb710ac3a411c0fd601
---
M server/eventlogging/handlers.py
M server/eventlogging/jrm.py
M server/eventlogging/utils.py
M server/tests/test_jrm.py
4 files changed, 82 insertions(+), 11 deletions(-)
Approvals:
QChris: Looks good to me, approved
diff --git a/server/eventlogging/handlers.py b/server/eventlogging/handlers.py
index 6e4bac0..d8de3bd 100644
--- a/server/eventlogging/handlers.py
+++ b/server/eventlogging/handlers.py
@@ -93,7 +93,11 @@
interval=2, target=store_sql_events, args=(meta, events))
worker.start()
- while 1:
+ # is_alive will be false if an exception
+ # was thrown on the worker thread
+ # main process will exit and will be re-started
+ # by upstart
+ while worker.is_alive():
event = (yield)
events.append(event)
diff --git a/server/eventlogging/jrm.py b/server/eventlogging/jrm.py
index 4f25bee..0ff0830 100644
--- a/server/eventlogging/jrm.py
+++ b/server/eventlogging/jrm.py
@@ -15,11 +15,11 @@
import sqlalchemy
-from .schema import get_schema, get_scid
+from .schema import get_schema
from .compat import items
-__all__ = ('store_sql_events',)
+__all__ = ('store_sql_events', 'flatten', 'key_func')
#: Format string for :func:`datetime.datetime.strptime` for MediaWiki
@@ -194,10 +194,20 @@
insert.execute()
+def key_func(event):
+ """Sort / group function that batches events that share the same
+ SCID and set of fields. We need to group together events that
+ belong to the same schema AND that have the same set of fields.
+ Please note that schemas allow for optional fields that might/might
+ not have been included in the event.
+ """
+ return (event['schema'], event['revision']), tuple(sorted(event))
+
+
def store_sql_events(meta, events, replace=False):
"""Store events in the database."""
- queue = [events.pop() for _ in range(len(events))]
- queue.sort(key=get_scid)
+ queue = [flatten(events.pop()) for _ in range(len(events))]
+ queue.sort(key=key_func)
if (getattr(meta.bind.dialect, 'supports_multivalues_insert', False)
or getattr(meta.bind.dialect, 'supports_multirow_insert', False)):
@@ -205,7 +215,7 @@
else:
insert = _insert_sequential
- for scid, events in itertools.groupby(queue, get_scid):
+ for (scid, _), events in itertools.groupby(queue, key_func):
prepared_events = [prepare(event) for event in events]
table = get_table(meta, scid)
insert(table, prepared_events, replace)
@@ -225,7 +235,6 @@
def prepare(event):
"""Prepare an event for insertion into the database."""
- event = flatten(event)
for prop in NO_DB_PROPERTIES:
event.pop(prop, None)
return event
diff --git a/server/eventlogging/utils.py b/server/eventlogging/utils.py
index 9c7e02c..cc530ed 100644
--- a/server/eventlogging/utils.py
+++ b/server/eventlogging/utils.py
@@ -7,6 +7,8 @@
a particular function.
"""
+from __future__ import unicode_literals
+
import threading
@@ -22,7 +24,7 @@
super(PeriodicThread, self).__init__(*args, **kwargs)
def run(self):
- self.ready.clear()
- self.ready.wait(self.interval)
- self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
- self.run()
+ while 1:
+ self.ready.clear()
+ self.ready.wait(self.interval)
+ self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
diff --git a/server/tests/test_jrm.py b/server/tests/test_jrm.py
index b3bfcba..ea7f062 100644
--- a/server/tests/test_jrm.py
+++ b/server/tests/test_jrm.py
@@ -13,6 +13,7 @@
import eventlogging
import sqlalchemy
+import itertools
from sqlalchemy.sql import select
from .fixtures import DatabaseTestMixin, TEST_SCHEMA_SCID
@@ -126,3 +127,58 @@
No exception is raised"""
event_list = []
eventlogging.store_sql_events(self.meta, event_list)
+
+ def test_grouping_of_events_happy_case(self):
+ """Events belonging to the same schema with the same
+ set of fields are gropued together """
+ another_event = next(self.event_generator)
+
+ event_list = [another_event, self.event]
+
+ queue = [eventlogging.flatten(event_list.pop())
+ for _ in range(len(event_list))]
+ queue.sort(key=eventlogging.key_func)
+
+ uniquekeys = []
+ for k, events in itertools.groupby(queue, eventlogging.key_func):
+ uniquekeys.append(k)
+ # we should have stored one key as events can be batched together
+ self.assertEquals(len(uniquekeys), 1)
+
+ def test_grouping_of_events_separately(self):
+ """Events belonging to the same schema with a different
+ set of optional fields are grouped separately
+ This might introspect the code too much, but
+ how else can we test?"""
+ another_event = next(self.event_generator)
+ # clientValidated is an optional field, remove it
+ del another_event['clientValidated']
+
+ event_list = [another_event, self.event]
+
+ queue = [eventlogging.flatten(event_list.pop())
+ for _ in range(len(event_list))]
+ queue.sort(key=eventlogging.key_func)
+
+ uniquekeys = []
+ for k, events in itertools.groupby(queue, eventlogging.key_func):
+ uniquekeys.append(k)
+ # we should have stored two keys as events cannot be grouped together
+ self.assertEquals(len(uniquekeys), 2)
+
+ def test_insert_events_with_different_set_of_optional_fields(self):
+ """Events belonging to the same schema with a different
+ set of optional fields are inserted correctly"""
+ another_event = next(self.event_generator)
+ # clientValidated is an optional field, remove it
+ del another_event['clientValidated']
+ # ensure both events get inserted?
+ event_list = [another_event, self.event]
+ eventlogging.store_sql_events(self.meta, event_list)
+ table = self.meta.tables['TestSchema_123']
+ # is the table on the db and does it have the right data?
+ s = select([table])
+ results = self.engine.execute(s)
+ # the number of records in table must be the list size
+ rows = results.fetchall()
+ self.assertEquals(len(rows), 2)
--
To view, visit https://gerrit.wikimedia.org/r/174485
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I0f160ef25d05f6b218b97cb710ac3a411c0fd601
Gerrit-PatchSet: 3
Gerrit-Project: mediawiki/extensions/EventLogging
Gerrit-Branch: master
Gerrit-Owner: Ori.livneh <[email protected]>
Gerrit-Reviewer: Nuria <[email protected]>
Gerrit-Reviewer: Ori.livneh <[email protected]>
Gerrit-Reviewer: QChris <[email protected]>
Gerrit-Reviewer: jenkins-bot <>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
