buffers: improve process_pending performance by making redis queries in parallel (#4597)

setup.py

@@ -141,7 +141,7 @@ install_requires = [
     'ua-parser>=0.6.1,<0.8.0',
     'urllib3>=1.14,<1.17',
     'uwsgi>2.0.0,<2.1.0',
-    'rb>=1.5.0,<2.0.0',
+    'rb>=1.6.0,<2.0.0',
     'qrcode>=5.2.2,<6.0.0',
     'python-u2flib-server>=4.0.1,<4.1.0',
 ]

src/sentry/buffer/redis.py

@@ -96,19 +96,19 @@ class RedisBuffer(Buffer):
 
         try:
             keycount = 0
-            for host_id in six.iterkeys(self.cluster.hosts):
-                conn = self.cluster.get_local_client(host_id)
-                keys = conn.zrange(self.pending_key, 0, -1)
-                if not keys:
-                    continue
-                for key in keys:
-                    keycount += 1
-                    process_incr.apply_async(kwargs={
-                        'key': key,
-                    })
-                pipe = conn.pipeline()
-                pipe.zrem(self.pending_key, *keys)
-                pipe.execute()
+            with self.cluster.all() as conn:
+                results = conn.zrange(self.pending_key, 0, -1)
+
+            with self.cluster.all() as conn:
+                for host_id, keys in six.iteritems(results.value):
+                    if not keys:
+                        continue
+                    keycount += len(keys)
+                    for key in keys:
+                        process_incr.apply_async(kwargs={
+                            'key': key,
+                        })
+                    conn.target([host_id]).zrem(self.pending_key, *keys)
             metrics.timing('buffer.pending-size', keycount)
         finally:
             client.delete(lock_key)