Merge branch 'kpn/55/metrics_KeyError' into 'master'

fix: metrics KeyError

See merge request !187

requirements/requirements.txt

@@ -2,7 +2,6 @@
 aiohttp[speedups]
 iso8601
 motor==2.4.0
-aiotask_context
 prozorro-metrics~=1.4
 prozorro-tools~=0.9
 ujson

src/mongodb_mirror/mirror.py

@@ -7,7 +7,6 @@ import pymongo
 import os
 import iso8601
 from aiohttp import web, WSCloseCode
-from aiotask_context import task_factory
 from diskcache import JSONDisk
 
 from bson import ObjectId
@@ -137,7 +136,8 @@ class MongoDBMirror:
             LOG.exception(msg)
             raise CriticalError(msg)
         LOG.info("Cursor closed. exiting")
-        clients_lag.remove(client.addr)
+        if client.addr in clients_lag._metrics:
+            clients_lag.remove(client.addr)
 
     async def get_available_cursor(self, timestamp):
         oplog_cursor = self._oplog.find(
@@ -372,17 +372,12 @@ class BaseApp:
     def __init__(self):
         self.__common_routes = [web.get('/ping', ping)]
         self.__common_middlewares = [request_id_middleware]
-        self._init_loop()
         self._configure_logging()
         self._load_auth()
 
     def _configure_logging(self):
         logger.configure_logging()
 
-    def _init_loop(self):
-        loop = asyncio.get_event_loop()
-        loop.set_task_factory(task_factory)
-
     def _load_auth(self):
         """
         Overwrite this method to load particular auth file

test/unit/test_mirror.py

@@ -94,8 +94,7 @@ class TestMongoDBMirror:
     @patch('mongodb_mirror.mirror.MongoDBMirror._serialize_data', return_value={'test': 'test'})
     @patch('mongodb_mirror.mirror.MongoDBMirror.get_available_cursor')
     @patch('mongodb_mirror.mirror.MongoDBMirror._get_oldest_oplog_ts', new_callable=AsyncMock)
-    @patch('mongodb_mirror.mirror.prometheus_client.Gauge.remove')
-    async def test_sync_client_noop_operation(self, mock_gauge_remove, mock_get_oldest_oplog_ts,
+    async def test_sync_client_noop_operation(self, mock_get_oldest_oplog_ts,
                                               mock_get_available_cursor, mock_serialize_data):
         entry_data = {
             'op': MongoOperation.NOOP,
@@ -114,14 +113,12 @@ class TestMongoDBMirror:
         mock_get_available_cursor.assert_called_once_with(self.current_ts)
         mock_serialize_data.assert_called_once_with(self.test_data)
         mock_client.noop.assert_called_once_with([self.current_ts.time, self.current_ts.inc], self.test_data)
-        mock_gauge_remove.assert_called_once()
 
     @pytest.mark.parametrize('operation', [MongoOperation.INSERT, MongoOperation.UPDATE])
     @patch('mongodb_mirror.mirror.MongoDBMirror._get_serialized_document', return_value={'test': 'test'})
     @patch('mongodb_mirror.mirror.MongoDBMirror.get_available_cursor')
     @patch('mongodb_mirror.mirror.MongoDBMirror._get_oldest_oplog_ts', new_callable=AsyncMock)
-    @patch('mongodb_mirror.mirror.prometheus_client.Gauge.remove')
-    async def test_sync_client_update_and_insert_operation(self, mock_gauge_remove, mock_get_oldest_oplog_ts,
+    async def test_sync_client_update_and_insert_operation(self, mock_get_oldest_oplog_ts,
                                                            mock_get_available_cursor, mock_serialize_document,
                                                            operation):
         entry_data = {
@@ -141,13 +138,11 @@ class TestMongoDBMirror:
         mock_get_available_cursor.assert_called_once_with(self.current_ts)
         mock_serialize_document.assert_called_once_with(self.test_data, 'procedures', self.current_ts)
         mock_client.upsert.assert_called_once_with([self.current_ts.time, self.current_ts.inc], self.test_data)
-        mock_gauge_remove.assert_called_once()
 
     @patch('mongodb_mirror.mirror.MongoDBMirror._serialize_data', return_value={'test': 'test'})
     @patch('mongodb_mirror.mirror.MongoDBMirror.get_available_cursor')
     @patch('mongodb_mirror.mirror.MongoDBMirror._get_oldest_oplog_ts', new_callable=AsyncMock)
-    @patch('mongodb_mirror.mirror.prometheus_client.Gauge.remove')
-    async def test_sync_client_delete_operation(self, mock_gauge_remove, mock_get_oldest_oplog_ts,
+    async def test_sync_client_delete_operation(self, mock_get_oldest_oplog_ts,
                                                 mock_get_available_cursor, mock_serialize_data):
         entry_data = {
             'op': MongoOperation.DELETE,
@@ -166,15 +161,13 @@ class TestMongoDBMirror:
         mock_get_available_cursor.assert_called_once_with(self.current_ts)
         mock_serialize_data.assert_called_once_with(self.test_data)
         mock_client.delete.assert_called_once_with([self.current_ts.time, self.current_ts.inc], self.test_data)
-        mock_gauge_remove.assert_called_once()
 
     @pytest.mark.parametrize('exception_type', (TypeError, ValueError, AttributeError, KeyError))
-    @patch('mongodb_mirror.mirror.prometheus_client.Gauge.remove')
     @patch('mongodb_mirror.mirror.MongoDBMirror._serialize_data', return_value={'test': 'test'})
     @patch('mongodb_mirror.mirror.MongoDBMirror.get_available_cursor')
     @patch('mongodb_mirror.mirror.MongoDBMirror._get_oldest_oplog_ts', new_callable=AsyncMock)
     async def test_sync_client_with_operations_fail(self, mock_get_oldest_oplog_ts, mock_get_available_cursor,
-                                                    mock_serialize_data, mock_gauge_remove, exception_type):
+                                                    mock_serialize_data, exception_type):
         data_with_obj_id = {**self.test_data, **{"_id": "test_id"}}
         entry_data = {
             'op': MongoOperation.NOOP,
@@ -201,7 +194,6 @@ class TestMongoDBMirror:
             object_id=data_with_obj_id.get("_id"),
             error_message=f'Critical error on trying to serialize entry {data_with_obj_id.get("_id")}')
         mock_serialize_data.assert_called_once_with(data_with_obj_id)
-        mock_gauge_remove.assert_called_once()
 
     def test_serialize_data(self):
         test_id = ObjectId('507f1f77bcf86cd799439011')