tileradotorg/synapse
1
0
Fork 0
forked from MirrorHub/synapse
Code Pull requests Activity

Clean up startup for the pusher (#6558)

Browse source 
* Remove redundant python2 support code

`str.decode()` doesn't exist on python3, so presumably this code was doing
nothing

* Filter out pushers with corrupt data

When we get a row with unparsable json, drop the row, rather than returning a
row with null `data`, which will then cause an explosion later on.

* Improve logging when we can't start a pusher

Log the ID to help us understand the problem

* Make email pusher setup more robust

We know we'll have a `data` member, since that comes from the database. What we
*don't* know is if that is a dict, and if that has a `brand` member, and if
that member is a string.
This commit is contained in:
Richard van der Hoff 2019-12-18 14:26:58 +00:00 committed by GitHub
parent 7963ca83cb
commit d6752ce5da
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 44 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/6558.misc Normal file
View file

@ -0,0 +1 @@
Clean up logs from the push notifier at startup.

12
synapse/push/pusher.py
View file

@ -80,9 +80,11 @@ class PusherFactory(object):
return EmailPusher(self.hs, pusherdict, mailer)
def _app_name_from_pusherdict(self, pusherdict):
if "data" in pusherdict and "brand" in pusherdict["data"]:
app_name = pusherdict["data"]["brand"]
else:
app_name = self.config.email_app_name
data = pusherdict["data"]
return app_name
if isinstance(data, dict):
brand = data.get("brand")
if isinstance(brand, str):
return brand
return self.config.email_app_name

10
synapse/push/pusherpool.py
View file

@ -232,7 +232,6 @@ class PusherPool:
Deferred
"""
pushers = yield self.store.get_all_pushers()
logger.info("Starting %d pushers", len(pushers))
# Stagger starting up the pushers so we don't completely drown the
# process on start up.
@ -245,7 +244,7 @@ class PusherPool:
"""Start the given pusher
Args:
pusherdict (dict):
pusherdict (dict): dict with the values pulled from the db table
Returns:
Deferred[EmailPusher|HttpPusher]
@ -254,7 +253,8 @@ class PusherPool:
p = self.pusher_factory.create_pusher(pusherdict)
except PusherConfigException as e:
logger.warning(
"Pusher incorrectly configured user=%s, appid=%s, pushkey=%s: %s",
"Pusher incorrectly configured id=%i, user=%s, appid=%s, pushkey=%s: %s",
pusherdict["id"],
pusherdict.get("user_name"),
pusherdict.get("app_id"),
pusherdict.get("pushkey"),
@ -262,7 +262,9 @@ class PusherPool:
)
return
except Exception:
logger.exception("Couldn't start a pusher: caught Exception")
logger.exception(
"Couldn't start pusher id %i: caught Exception", pusherdict["id"],
)
return
if not p:

31
synapse/rest/client/v1/pusher.py
View file

@ -28,6 +28,17 @@ from synapse.rest.client.v2_alpha._base import client_patterns
logger = logging.getLogger(__name__)
ALLOWED_KEYS = {
"app_display_name",
"app_id",
"data",
"device_display_name",
"kind",
"lang",
"profile_tag",
"pushkey",
}
class PushersRestServlet(RestServlet):
PATTERNS = client_patterns("/pushers$", v1=True)
@ -43,23 +54,11 @@ class PushersRestServlet(RestServlet):
pushers = await self.hs.get_datastore().get_pushers_by_user_id(user.to_string())
allowed_keys = [
"app_display_name",
"app_id",
"data",
"device_display_name",
"kind",
"lang",
"profile_tag",
"pushkey",
]
filtered_pushers = list(
{k: v for k, v in p.items() if k in ALLOWED_KEYS} for p in pushers
)
for p in pushers:
for k, v in list(p.items()):
if k not in allowed_keys:
del p[k]
return 200, {"pushers": pushers}
return 200, {"pushers": filtered_pushers}
def on_OPTIONS(self, _):
return 200, {}

25
synapse/storage/data_stores/main/pusher.py
View file

@ -15,8 +15,7 @@
# limitations under the License.
import logging
import six
from typing import Iterable, Iterator
from canonicaljson import encode_canonical_json, json
@ -27,21 +26,16 @@ from synapse.util.caches.descriptors import cachedInlineCallbacks, cachedList
logger = logging.getLogger(__name__)
if six.PY2:
db_binary_type = six.moves.builtins.buffer
else:
db_binary_type = memoryview
class PusherWorkerStore(SQLBaseStore):
def _decode_pushers_rows(self, rows):
def _decode_pushers_rows(self, rows: Iterable[dict]) -> Iterator[dict]:
"""JSON-decode the data in the rows returned from the `pushers` table
Drops any rows whose data cannot be decoded
"""
for r in rows:
dataJson = r["data"]
r["data"] = None
try:
if isinstance(dataJson, db_binary_type):
dataJson = str(dataJson).decode("UTF8")
r["data"] = json.loads(dataJson)
except Exception as e:
logger.warning(
@ -50,12 +44,9 @@ class PusherWorkerStore(SQLBaseStore):
dataJson,
e.args[0],
)
pass
continue
if isinstance(r["pushkey"], db_binary_type):
r["pushkey"] = str(r["pushkey"]).decode("UTF8")
return rows
yield r
@defer.inlineCallbacks
def user_has_pusher(self, user_id):

3
tests/push/test_email.py
View file

@ -165,6 +165,7 @@ class EmailPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=self.user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
last_stream_ordering = pushers[0]["last_stream_ordering"]
@ -175,6 +176,7 @@ class EmailPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=self.user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertEqual(last_stream_ordering, pushers[0]["last_stream_ordering"])
@ -192,5 +194,6 @@ class EmailPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=self.user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertTrue(pushers[0]["last_stream_ordering"] > last_stream_ordering)

4
tests/push/test_http.py
View file

@ -104,6 +104,7 @@ class HTTPPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
last_stream_ordering = pushers[0]["last_stream_ordering"]
@ -114,6 +115,7 @@ class HTTPPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertEqual(last_stream_ordering, pushers[0]["last_stream_ordering"])
@ -132,6 +134,7 @@ class HTTPPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertTrue(pushers[0]["last_stream_ordering"] > last_stream_ordering)
last_stream_ordering = pushers[0]["last_stream_ordering"]
@ -151,5 +154,6 @@ class HTTPPusherTests(HomeserverTestCase):
pushers = self.get_success(
self.hs.get_datastore().get_pushers_by(dict(user_name=user_id))
)
pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertTrue(pushers[0]["last_stream_ordering"] > last_stream_ordering)