Implement configurable stats reporting

SYN-287

This requires that HS owners either opt in or out of stats reporting.

When --generate-config is passed, --report-stats must be specified
If an already-generated config is used, and doesn't have the
report_stats key, it is requested to be set.

4 files changed, 213 insertions, 54 deletions

diff --git a/tests/storage/event_injector.py b/tests/storage/event_injector.py
new file mode 100644
index 0000000000..42bd8928bd
--- /dev/null
+++ b/tests/storage/event_injector.py
@@ -0,0 +1,81 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from tests import unittest
+from twisted.internet import defer
+
+from synapse.api.constants import EventTypes, Membership
+from synapse.types import UserID, RoomID
+
+from tests.utils import setup_test_homeserver
+
+from mock import Mock
+
+
+class EventInjector:
+    def __init__(self, hs):
+        self.hs = hs
+        self.store = hs.get_datastore()
+        self.message_handler = hs.get_handlers().message_handler
+        self.event_builder_factory = hs.get_event_builder_factory()
+
+    @defer.inlineCallbacks
+    def create_room(self, room):
+        builder = self.event_builder_factory.new({
+            "type": EventTypes.Create,
+            "room_id": room.to_string(),
+            "content": {},
+        })
+
+        event, context = yield self.message_handler._create_new_client_event(
+            builder
+        )
+
+        yield self.store.persist_event(event, context)
+
+    @defer.inlineCallbacks
+    def inject_room_member(self, room, user, membership):
+        builder = self.event_builder_factory.new({
+            "type": EventTypes.Member,
+            "sender": user.to_string(),
+            "state_key": user.to_string(),
+            "room_id": room.to_string(),
+            "content": {"membership": membership},
+        })
+
+        event, context = yield self.message_handler._create_new_client_event(
+            builder
+        )
+
+        yield self.store.persist_event(event, context)
+
+        defer.returnValue(event)
+
+    @defer.inlineCallbacks
+    def inject_message(self, room, user, body):
+        builder = self.event_builder_factory.new({
+            "type": EventTypes.Message,
+            "sender": user.to_string(),
+            "state_key": user.to_string(),
+            "room_id": room.to_string(),
+            "content": {"body": body, "msgtype": u"message"},
+        })
+
+        event, context = yield self.message_handler._create_new_client_event(
+            builder
+        )
+
+        yield self.store.persist_event(event, context)
diff --git a/tests/storage/test_events.py b/tests/storage/test_events.py
new file mode 100644
index 0000000000..313013009e
--- /dev/null
+++ b/tests/storage/test_events.py
@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import uuid
+from mock.mock import Mock
+from synapse.types import RoomID, UserID
+
+from tests import unittest
+from twisted.internet import defer
+from tests.storage.event_injector import EventInjector
+
+from tests.utils import setup_test_homeserver
+
+
+class EventsStoreTestCase(unittest.TestCase):
+
+    @defer.inlineCallbacks
+    def setUp(self):
+        self.hs = yield setup_test_homeserver(
+            resource_for_federation=Mock(),
+            http_client=None,
+        )
+        self.store = self.hs.get_datastore()
+        self.db_pool = self.hs.get_db_pool()
+        self.message_handler = self.hs.get_handlers().message_handler
+        self.event_injector = EventInjector(self.hs)
+
+    @defer.inlineCallbacks
+    def test_count_daily_messages(self):
+        self.db_pool.runQuery("DELETE FROM stats_reporting")
+
+        self.hs.clock.now = 100
+
+        # Never reported before, and nothing which could be reported
+        count = yield self.store.count_daily_messages()
+        self.assertIsNone(count)
+        count = yield self.db_pool.runQuery("SELECT COUNT(*) FROM stats_reporting")
+        self.assertEqual([(0,)], count)
+
+        # Create something to report
+        room = RoomID.from_string("!abc123:test")
+        user = UserID.from_string("@raccoonlover:test")
+        yield self.event_injector.create_room(room)
+
+        self.base_event = yield self._get_last_stream_token()
+
+        yield self.event_injector.inject_message(room, user, "Raccoons are really cute")
+
+        # Never reported before, something could be reported, but isn't because
+        # it isn't old enough.
+        count = yield self.store.count_daily_messages()
+        self.assertIsNone(count)
+        self._assert_stats_reporting(1, self.hs.clock.now)
+
+        # Already reported yesterday, two new events from today.
+        yield self.event_injector.inject_message(room, user, "Yeah they are!")
+        yield self.event_injector.inject_message(room, user, "Incredibly!")
+        self.hs.clock.now += 60 * 60 * 24
+        count = yield self.store.count_daily_messages()
+        self.assertEqual(2, count)  # 2 since yesterday
+        self._assert_stats_reporting(3, self.hs.clock.now)  # 3 ever
+
+        # Last reported too recently.
+        yield self.event_injector.inject_message(room, user, "Who could disagree?")
+        self.hs.clock.now += 60 * 60 * 22
+        count = yield self.store.count_daily_messages()
+        self.assertIsNone(count)
+        self._assert_stats_reporting(4, self.hs.clock.now)
+
+        # Last reported too long ago
+        yield self.event_injector.inject_message(room, user, "No one.")
+        self.hs.clock.now += 60 * 60 * 26
+        count = yield self.store.count_daily_messages()
+        self.assertIsNone(count)
+        self._assert_stats_reporting(5, self.hs.clock.now)
+
+        # And now let's actually report something
+        yield self.event_injector.inject_message(room, user, "Indeed.")
+        yield self.event_injector.inject_message(room, user, "Indeed.")
+        yield self.event_injector.inject_message(room, user, "Indeed.")
+        # A little over 24 hours is fine :)
+        self.hs.clock.now += (60 * 60 * 24) + 50
+        count = yield self.store.count_daily_messages()
+        self.assertEqual(3, count)
+        self._assert_stats_reporting(8, self.hs.clock.now)
+
+    @defer.inlineCallbacks
+    def _get_last_stream_token(self):
+        rows = yield self.db_pool.runQuery(
+            "SELECT stream_ordering"
+            " FROM events"
+            " ORDER BY stream_ordering DESC"
+            " LIMIT 1"
+        )
+        if not rows:
+            defer.returnValue(0)
+        else:
+            defer.returnValue(rows[0][0])
+
+    @defer.inlineCallbacks
+    def _assert_stats_reporting(self, messages, time):
+        rows = yield self.db_pool.runQuery(
+            "SELECT reported_stream_token, reported_time FROM stats_reporting"
+        )
+        self.assertEqual([(self.base_event + messages, time,)], rows)
diff --git a/tests/storage/test_room.py b/tests/storage/test_room.py
index ab7625a3ca..caffce64e3 100644
--- a/tests/storage/test_room.py
+++ b/tests/storage/test_room.py
@@ -85,7 +85,7 @@ class RoomEventsStoreTestCase(unittest.TestCase):
         # Room events need the full datastore, for persist_event() and
         # get_room_state()
         self.store = hs.get_datastore()
-        self.event_factory = hs.get_event_factory();
+        self.event_factory = hs.get_event_factory()
 
         self.room = RoomID.from_string("!abcde:test")
 
diff --git a/tests/storage/test_stream.py b/tests/storage/test_stream.py
index 0c9b89d765..a658a789aa 100644
--- a/tests/storage/test_stream.py
+++ b/tests/storage/test_stream.py
@@ -19,6 +19,7 @@ from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
 from synapse.types import UserID, RoomID
+from tests.storage.event_injector import EventInjector
 
 from tests.utils import setup_test_homeserver
 
@@ -36,6 +37,7 @@ class StreamStoreTestCase(unittest.TestCase):
 
         self.store = hs.get_datastore()
         self.event_builder_factory = hs.get_event_builder_factory()
+        self.event_injector = EventInjector(hs)
         self.handlers = hs.get_handlers()
         self.message_handler = self.handlers.message_handler
 
@@ -45,60 +47,20 @@ class StreamStoreTestCase(unittest.TestCase):
         self.room1 = RoomID.from_string("!abc123:test")
         self.room2 = RoomID.from_string("!xyx987:test")
 
-        self.depth = 1
-
-    @defer.inlineCallbacks
-    def inject_room_member(self, room, user, membership):
-        self.depth += 1
-
-        builder = self.event_builder_factory.new({
-            "type": EventTypes.Member,
-            "sender": user.to_string(),
-            "state_key": user.to_string(),
-            "room_id": room.to_string(),
-            "content": {"membership": membership},
-        })
-
-        event, context = yield self.message_handler._create_new_client_event(
-            builder
-        )
-
-        yield self.store.persist_event(event, context)
-
-        defer.returnValue(event)
-
-    @defer.inlineCallbacks
-    def inject_message(self, room, user, body):
-        self.depth += 1
-
-        builder = self.event_builder_factory.new({
-            "type": EventTypes.Message,
-            "sender": user.to_string(),
-            "state_key": user.to_string(),
-            "room_id": room.to_string(),
-            "content": {"body": body, "msgtype": u"message"},
-        })
-
-        event, context = yield self.message_handler._create_new_client_event(
-            builder
-        )
-
-        yield self.store.persist_event(event, context)
-
     @defer.inlineCallbacks
     def test_event_stream_get_other(self):
         # Both bob and alice joins the room
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_alice, Membership.JOIN
         )
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_bob, Membership.JOIN
         )
 
         # Initial stream key:
         start = yield self.store.get_room_events_max_id()
 
-        yield self.inject_message(self.room1, self.u_alice, u"test")
+        yield self.event_injector.inject_message(self.room1, self.u_alice, u"test")
 
         end = yield self.store.get_room_events_max_id()
 
@@ -125,17 +87,17 @@ class StreamStoreTestCase(unittest.TestCase):
     @defer.inlineCallbacks
     def test_event_stream_get_own(self):
         # Both bob and alice joins the room
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_alice, Membership.JOIN
         )
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_bob, Membership.JOIN
         )
 
         # Initial stream key:
         start = yield self.store.get_room_events_max_id()
 
-        yield self.inject_message(self.room1, self.u_alice, u"test")
+        yield self.event_injector.inject_message(self.room1, self.u_alice, u"test")
 
         end = yield self.store.get_room_events_max_id()
 
@@ -162,22 +124,22 @@ class StreamStoreTestCase(unittest.TestCase):
     @defer.inlineCallbacks
     def test_event_stream_join_leave(self):
         # Both bob and alice joins the room
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_alice, Membership.JOIN
         )
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_bob, Membership.JOIN
         )
 
         # Then bob leaves again.
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_bob, Membership.LEAVE
         )
 
         # Initial stream key:
         start = yield self.store.get_room_events_max_id()
 
-        yield self.inject_message(self.room1, self.u_alice, u"test")
+        yield self.event_injector.inject_message(self.room1, self.u_alice, u"test")
 
         end = yield self.store.get_room_events_max_id()
 
@@ -193,17 +155,17 @@ class StreamStoreTestCase(unittest.TestCase):
 
     @defer.inlineCallbacks
     def test_event_stream_prev_content(self):
-        yield self.inject_room_member(
+        yield self.event_injector.inject_room_member(
             self.room1, self.u_bob, Membership.JOIN
         )
 
-        event1 = yield self.inject_room_member(
+        event1 = yield self.event_injector.inject_room_member(
             self.room1, self.u_alice, Membership.JOIN
         )
 
         start = yield self.store.get_room_events_max_id()
 
-        event2 = yield self.inject_room_member(
+        event2 = yield self.event_injector.inject_room_member(
             self.room1, self.u_alice, Membership.JOIN,
         )