fix: update connection counter instead of inc/dec (#237)

* fix: update connection counter instead of inc/dec

Author: rpurdel

skynet/modules/monitoring.py

@@ -138,28 +138,14 @@
     metrics.requests(metric_namespace=PROMETHEUS_NAMESPACE, metric_subsystem=PROMETHEUS_SUMMARIES_SUBSYSTEM),
 )
 
-ws_connection_count = 0
 
+async def update_ws_conn_count(num: int):
+    CONNECTIONS_METRIC.set(num)
+    set_ts_stress_level(num)
 
-def inc_ws_conn_count():
-    global ws_connection_count
-    ws_connection_count += 1
-    CONNECTIONS_METRIC.set(ws_connection_count)
-    TRANSCRIBE_CONNECTIONS_COUNTER.inc()
-    set_ts_stress_level()
 
-
-def dec_ws_conn_count():
-    global ws_connection_count
-    if ws_connection_count > 0:
-        ws_connection_count -= 1
-    CONNECTIONS_METRIC.set(ws_connection_count)
-    set_ts_stress_level()
-
-
-def set_ts_stress_level():
-    global ws_connection_count
-    if ws_connection_count > 0:
-        TRANSCRIBE_STRESS_LEVEL_METRIC.set(ws_connection_count / whisper_max_connections)
+def set_ts_stress_level(conns: int):
+    if conns > 0:
+        TRANSCRIBE_STRESS_LEVEL_METRIC.set(conns / whisper_max_connections)
     else:
         TRANSCRIBE_STRESS_LEVEL_METRIC.set(0.0)

skynet/modules/stt/streaming_whisper/connection_manager.py

@@ -6,7 +6,7 @@
 from skynet.auth.jwt import authorize
 from skynet.env import bypass_auth, whisper_flush_interval
 from skynet.logs import get_logger
-from skynet.modules.monitoring import dec_ws_conn_count, inc_ws_conn_count
+from skynet.modules.monitoring import update_ws_conn_count
 from skynet.modules.stt.streaming_whisper.meeting_connection import MeetingConnection
 from skynet.modules.stt.streaming_whisper.utils import utils
 
@@ -34,9 +34,9 @@ async def connect(self, websocket: WebSocket, meeting_id: str, auth_token: str |
         if self.flush_audio_task is None:
             loop = asyncio.get_running_loop()
             self.flush_audio_task = loop.create_task(self.flush_working_audio_worker())
-        inc_ws_conn_count()
-        log.info(f'Meeting with id {meeting_id} started. Ongoing meetings {len(self.connections)}')
-
+        current_connections = len(self.connections)
+        log.info(f'Meeting with id {meeting_id} started. Ongoing meetings {current_connections}')
+        await update_ws_conn_count(current_connections)
         return connection
 
     async def process(self, connection: MeetingConnection, chunk: bytes, chunk_timestamp: int):
@@ -73,7 +73,9 @@ async def disconnect(self, connection: MeetingConnection, already_closed=False):
         else:
             # mark connection as disconnected
             connection.disconnect()
-        dec_ws_conn_count()
+        remaining_connections = len(self.connections)
+        log.info(f'Disconnected meeting {connection.meeting_id}, remaining connections {remaining_connections}')
+        await update_ws_conn_count(remaining_connections)
 
     async def flush_working_audio_worker(self):
         """