[web_tests][resultdb] use requests instead of urllib2

urllib2 doesn't support connection pooling. Therefore, each of sink()
invocation creates a new TCP connection established. It seems that
Linux and Mac establish TCP connections much quicker than Windows.
However, creation of TCP connections added a significant amount of
delays to blink web test runs in Windows bots.

This CL solves it by using requests, which uses urllib3 and supports
connection pooling.

R=robertma@chromium.org
CC=chanli@chromium.org,bpastene@chromium.org,nodir@chromium.org,zhaoyangli@chromium.org

Bug: 1141606
Change-Id: I5fc63894ed31867db1ae1b4eac5a44b5e0921c5c
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2520088
Commit-Queue: Scott Lee <ddoman@chromium.org>
Reviewed-by: Nodir Turakulov <nodir@chromium.org>
Reviewed-by: Robert Ma <robertma@chromium.org>
Cr-Commit-Position: refs/heads/master@{#824622}

third_party/blink/tools/blinkpy/web_tests/controllers/manager.py

@@ -94,10 +94,10 @@ class Manager(object):
         self._finder = WebTestFinder(self._port, self._options)
         self._path_finder = PathFinder(port.host.filesystem)
 
-        sink = CreateTestResultSink(self._port)
+        self._sink = CreateTestResultSink(self._port)
         self._runner = WebTestRunner(self._options, self._port, self._printer,
                                      self._results_directory,
-                                     self._test_is_slow, sink)
+                                     self._test_is_slow, self._sink)
 
     def run(self, args):
         """Runs the tests and return a RunDetails object with the results."""
@@ -499,6 +499,9 @@ class Manager(object):
         sys.stderr.flush()
         _log.debug('Cleaning up port')
         self._port.clean_up_test_run()
+        if self._sink:
+            _log.debug('Closing sink')
+            self._sink.close()
 
     def _look_for_new_crash_logs(self, run_results, start_time):
         """Looks for and writes new crash logs, at the end of the test run.

third_party/blink/tools/blinkpy/web_tests/controllers/test_result_sink.py

@@ -15,7 +15,7 @@ section.
 
 import json
 import logging
-import urllib2
+import requests
 
 from blinkpy.web_tests.models.typ_types import ResultType
 
@@ -44,6 +44,10 @@ _result_type_to_sink_status = {
 }
 
 
+class TestResultSinkClosed(Exception):
+    """Raises if sink() is called over a closed TestResultSink instance."""
+
+
 def CreateTestResultSink(port):
     """Creates TestResultSink, if result_sink is present in LUCI_CONTEXT.
 
@@ -70,23 +74,21 @@ class TestResultSink(object):
 
     def __init__(self, port, sink_ctx):
         self._port = port
+        self.is_closed = False
         self._sink_ctx = sink_ctx
-        self._sink_url = (
+        self._url = (
             'http://%s/prpc/luci.resultsink.v1.Sink/ReportTestResults' %
             self._sink_ctx['address'])
+        self._session = requests.Session()
+        sink_headers = {
+            'Content-Type': 'application/json',
+            'Accept': 'application/json',
+            'Authorization': 'ResultSink %s' % self._sink_ctx['auth_token'],
+        }
+        self._session.headers.update(sink_headers)
 
     def _send(self, data):
-        req = urllib2.Request(
-            url=self._sink_url,
-            data=json.dumps(data),
-            headers={
-                'Content-Type': 'application/json',
-                'Accept': 'application/json',
-                'Authorization':
-                'ResultSink %s' % self._sink_ctx['auth_token'],
-            },
-        )
-        return urllib2.urlopen(req)
+        self._session.post(self._url, data=json.dumps(data)).raise_for_status()
 
     def _status(self, result):
         """Returns the TestStatus enum value corresponding to the result type.
@@ -152,9 +154,15 @@ class TestResultSink(object):
                 False, otherwise.
             result: The TestResult object to report.
         Exceptions:
-            urllib2.URLError, if there was a network connection error.
-            urllib2.HTTPError, if ResultSink responded an error for the request.
+            requests.exceptions.ConnectionError, if there was a network
+              connection error.
+            requests.exceptions.HTTPError, if ResultSink responded an error
+              for the request.
+            ResultSinkClosed, if sink.close() was called prior to sink().
         """
+        if self.is_closed:
+            raise TestResultSinkClosed('sink() cannot be called after close()')
+
         # The structure and member definitions of this dict can be found at
         # https://chromium.googlesource.com/infra/luci/luci-go/+/refs/heads/master/resultdb/proto/sink/v1/test_result.proto
         r = {
@@ -178,3 +186,12 @@ class TestResultSink(object):
             },
         }
         self._send({'testResults': [r]})
+
+    def close(self):
+        """Close closes all the active connections to SinkServer.
+
+        The sink object is no longer usable after being closed.
+        """
+        if not self.is_closed:
+            self.is_closed = True
+            self._session.close()

third_party/blink/tools/blinkpy/web_tests/controllers/test_result_sink_unittest.py

@@ -5,6 +5,7 @@
 import contextlib
 import json
 import mock
+import requests
 import sys
 import unittest
 from urlparse import urlparse
@@ -40,20 +41,29 @@ class TestCreateTestResultSink(TestResultSinkTestBase):
         self.luci_context(app={'foo': 'bar'})
         self.assertIsNone(CreateTestResultSink(self.port))
 
-    @mock.patch('urllib2.urlopen')
-    def test_with_result_sink_section(self, urlopen):
+    def test_auth_token(self):
         ctx = {'address': 'localhost:123', 'auth_token': 'secret'}
         self.luci_context(result_sink=ctx)
-        r = CreateTestResultSink(self.port)
-        self.assertIsNotNone(r)
-        r.sink(True, test_results.TestResult('test'))
-
-        urlopen.assert_called_once()
-        req = urlopen.call_args[0][0]
-        self.assertEqual(urlparse(req.get_full_url()).netloc, ctx['address'])
-        self.assertEqual(req.get_header('Authorization'),
+        rs = CreateTestResultSink(self.port)
+        self.assertIsNotNone(rs)
+        self.assertEqual(rs._session.headers['Authorization'],
                          'ResultSink ' + ctx['auth_token'])
 
+    def test_with_result_sink_section(self):
+        ctx = {'address': 'localhost:123', 'auth_token': 'secret'}
+        self.luci_context(result_sink=ctx)
+        rs = CreateTestResultSink(self.port)
+        self.assertIsNotNone(rs)
+
+        response = requests.Response()
+        response.status_code = 200
+        with mock.patch.object(rs._session, 'post',
+                               return_value=response) as m:
+            rs.sink(True, test_results.TestResult('test'))
+            self.assertTrue(m.called)
+            self.assertEqual(
+                urlparse(m.call_args[0][0]).netloc, ctx['address'])
+
 
 class TestResultSinkMessage(TestResultSinkTestBase):
     """Tests ResulkSink.sink."""
@@ -66,11 +76,11 @@ class TestResultSinkMessage(TestResultSinkTestBase):
 
         ctx = {'address': 'localhost:123', 'auth_token': 'super-secret'}
         self.luci_context(result_sink=ctx)
-        self.r = CreateTestResultSink(self.port)
+        self.rs = CreateTestResultSink(self.port)
 
     def sink(self, expected, test_result):
-        self.r.sink(expected, test_result)
-        self.mock_send.called_once()
+        self.rs.sink(expected, test_result)
+        self.assertTrue(self.mock_send.called)
         return self.mock_send.call_args[0][0]['testResults'][0]
 
     def test_sink(self):