feat: add requests tracking mixin | track dAuth requests

Author: toderian

extensions/business/dauth/dauth_manager.py

@@ -22,6 +22,7 @@
 """
 from extensions.business.mixins.node_tags_mixin import _NodeTagsMixin
 from naeural_core.business.default.web_app.supervisor_fast_api_web_app import SupervisorFastApiWebApp as BasePlugin
+from extensions.business.mixins.request_tracking_mixin import _RequestTrackingMixin
 from extensions.business.dauth.dauth_mixin import _DauthMixin
 
 __VER__ = '0.2.2'
@@ -36,6 +37,10 @@
   'DAUTH_VERBOSE' : False,
   'DAUTH_LOG_RESPONSE' : True,
 
+  'REQUESTS_CSTORE_HKEY': 'DAUTH_REQUESTS',
+  'REQUESTS_MAX_RECORDS': 2,
+  'REQUESTS_LOG_INTERVAL': 5 * 60,
+
   'SUPRESS_LOGS_AFTER_INTERVAL' : 300,
 
   # required ENV keys are defined in plugin template and should be added here  
@@ -83,7 +88,8 @@
 class DauthManagerPlugin(
   BasePlugin,
   _DauthMixin,
-  _NodeTagsMixin
+  _NodeTagsMixin,
+  _RequestTrackingMixin,
   ):
   """
   This plugin is the dAuth FastAPI web app that provides an endpoints for decentralized authentication.
@@ -103,10 +109,23 @@ def on_init(self):
     self.P("Started {} plugin on {} / {}\n - Auth keys: {}\n - Predefined keys: {}".format(
       self.__class__.__name__, my_address, my_eth_address,
       self.cfg_auth_env_keys, self.cfg_auth_predefined_keys)
-    )        
+    )
+    self._init_request_tracking()
     return
 
 
+  def on_request(self, request):
+    self._track_request(request)
+    return
+
+  def on_response(self, method, response):
+    self._track_response(method, response)
+    return
+
+  def process(self):
+    self._maybe_log_tracked_requests()
+    return
+
   def __get_current_epoch(self):
     """
     Get the current epoch of the node.

extensions/business/deeploy/deeploy_manager_api.py

@@ -10,6 +10,7 @@
 from .deeploy_mixin import _DeeployMixin
 from .deeploy_target_nodes_mixin import _DeeployTargetNodesMixin
 from extensions.business.mixins.node_tags_mixin import _NodeTagsMixin
+from extensions.business.mixins.request_tracking_mixin import _RequestTrackingMixin
 from .deeploy_const import (
   DEEPLOY_CREATE_REQUEST, DEEPLOY_CREATE_REQUEST_MULTI_PLUGIN, DEEPLOY_GET_APPS_REQUEST, DEEPLOY_DELETE_REQUEST,
   DEEPLOY_ERRORS, DEEPLOY_KEYS, DEEPLOY_SCALE_UP_JOB_WORKERS_REQUEST, DEEPLOY_STATUS, DEEPLOY_INSTANCE_COMMAND_REQUEST,
@@ -20,8 +21,6 @@
 
 from naeural_core.business.default.web_app.supervisor_fast_api_web_app import SupervisorFastApiWebApp as BasePlugin
 
-DEEPLOY_REQUESTS_CSTORE_HKEY = "DEEPLOY_REQUESTS"
-
 __VER__ = '0.6.0'
 
 
@@ -39,6 +38,8 @@
   'WARMUP_DELAY' : 300,
   'PIPELINES_CHECK_DELAY' : 300,
   'MIN_ETH_BALANCE' : 0.00005,
+
+  'REQUESTS_CSTORE_HKEY': 'DEEPLOY_REQUESTS',
   'REQUESTS_LOG_INTERVAL' : 5 * 60,
   'REQUESTS_MAX_RECORDS' : 2,
 
@@ -55,6 +56,7 @@ class DeeployManagerApiPlugin(
   _DeeployTargetNodesMixin,
   _NodeTagsMixin,
   _DeeployJobMixin,
+  _RequestTrackingMixin,
   ):
   """
   This plugin is the dAuth FastAPI web app that provides an endpoints for decentralized authentication.
@@ -86,68 +88,16 @@ def on_init(self):
         color='r', boxed=True
       )
       self.maybe_stop_tunnel_engine()
-    # Request tracking state
-    self.__recent_requests = self.deque(maxlen=self.cfg_requests_max_records)
-    self.__last_requests_log_time = 0
+    self._init_request_tracking()
     return
 
 
   def on_request(self, request):
-    """
-    Hook called when a new request arrives from the FastAPI side (monitor thread).
-    Captures minimal request metadata and writes the last N records to cstore.
-
-    Parameters
-    ----------
-    request : dict
-        Raw request payload pulled from the server queue.
-        Structure: {'id': str, 'value': tuple, 'profile': dict|None}
-    """
-    try:
-      value = request.get('value')
-      request_id = request.get('id')
-      endpoint = value[0] if isinstance(value, (list, tuple)) and len(value) > 0 else 'unknown'
-      record = {
-        'id': request_id,
-        'endpoint': endpoint,
-        'date_start': self.datetime.now(self.timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
-        'date_complete': None,
-      }
-      self.__recent_requests.append(record)
-      self._save_requests_to_cstore()
-    except Exception as e:
-      self.P(f"Error tracking request in cstore: {e}", color='r')
+    self._track_request(request)
     return
 
-
   def on_response(self, method, response):
-    """
-    Hook called before the response is sent back to the FastAPI side (main thread).
-
-    Parameters
-    ----------
-    method : str
-        The endpoint name that was called.
-    response : dict
-        Response dict with 'id' and 'value' keys.
-    """
-    try:
-      request_id = response.get('id')
-      for record in self.__recent_requests:
-        if record.get('id') == request_id:
-          record['date_complete'] = self.datetime.now(self.timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
-          self._save_requests_to_cstore()
-          break
-    except Exception as e:
-      self.P(f"Error tracking response in cstore: {e}", color='r')
-    return
-
-  def _save_requests_to_cstore(self):
-    self.chainstore_hset(
-      hkey=DEEPLOY_REQUESTS_CSTORE_HKEY,
-      key=self.ee_id,
-      value=list(self.__recent_requests),
-    )
+    self._track_response(method, response)
     return
 
 
@@ -1117,17 +1067,5 @@ def process(self):
         self.P(f"Error checking running pipelines: {e}", color='r')
       self.__last_pipelines_check_time = self.time()
 
-    # Periodic dump of all nodes' recent requests from cstore
-    if (self.time() - self.__last_requests_log_time) > self.cfg_requests_log_interval:
-      try:
-        all_requests = self.chainstore_hgetall(hkey=DEEPLOY_REQUESTS_CSTORE_HKEY)
-        if all_requests:
-          self.P(f"Deeploy requests across all nodes:\n{self.json_dumps(all_requests, indent=2)}")
-        else:
-          self.P("Deeploy requests across all nodes: no data")
-      except Exception as e:
-        self.P(f"Error dumping deeploy requests from cstore: {e}", color='r')
-      # end try
-      self.__last_requests_log_time = self.time()
-    # end if
+    self._maybe_log_tracked_requests()
     return

extensions/business/mixins/request_tracking_mixin.py

@@ -0,0 +1,107 @@
+DEFAULT_REQUESTS_MAX_RECORDS = 2
+DEFAULT_REQUESTS_LOG_INTERVAL = 5 * 60  # 300 seconds
+
+
+class _RequestTrackingMixin(object):
+  """
+  Mixin that adds chainstore-based request/response tracking to FastAPI plugins.
+
+  Opt-in: set REQUESTS_CSTORE_HKEY in plugin config to enable.
+  When not set (None), all methods are no-ops.
+
+  Config keys:
+    REQUESTS_CSTORE_HKEY : str or None  -- chainstore hash key (None = disabled)
+    REQUESTS_MAX_RECORDS : int          -- max recent requests in deque (default 2)
+    REQUESTS_LOG_INTERVAL : int         -- seconds between cross-node log dumps (default 300)
+  """
+
+  @property
+  def __rt_max_records(self):
+    val = getattr(self, 'cfg_requests_max_records', None)
+    if not isinstance(val, int) or val < 1:
+      return DEFAULT_REQUESTS_MAX_RECORDS
+    return val
+
+  @property
+  def __rt_log_interval(self):
+    val = getattr(self, 'cfg_requests_log_interval', None)
+    if not isinstance(val, (int, float)) or val <= 0:
+      return DEFAULT_REQUESTS_LOG_INTERVAL
+    return val
+
+  @property
+  def __rt_cstore_hkey(self):
+    return getattr(self, 'cfg_requests_cstore_hkey', None)
+
+  def _init_request_tracking(self):
+    """Call from on_init(). Initializes tracking state if enabled."""
+    self.__rt_recent_requests = None
+    self.__rt_last_log_time = 0
+    if self.__rt_cstore_hkey:
+      self.__rt_recent_requests = self.deque(maxlen=self.__rt_max_records)
+    return
+
+
+  def _track_request(self, request):
+    """Called from the request processing flow. Records request start."""
+    if self.__rt_recent_requests is None:
+      return
+    try:
+      value = request.get('value')
+      request_id = request.get('id')
+      endpoint = value[0] if isinstance(value, (list, tuple)) and len(value) > 0 else 'unknown'
+      record = {
+        'id': request_id,
+        'endpoint': endpoint,
+        'date_start': self.datetime.now(self.timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
+        'date_complete': None,
+      }
+      self.__rt_recent_requests.append(record)
+      self.__rt_save()
+    except Exception as e:
+      self.P(f"Error tracking request in cstore: {e}", color='r')
+    return
+
+
+  def _track_response(self, method, response):
+    """Called from the response processing flow. Stamps completion time."""
+    if self.__rt_recent_requests is None:
+      return
+    try:
+      request_id = response.get('id')
+      for record in self.__rt_recent_requests:
+        if record.get('id') == request_id:
+          record['date_complete'] = self.datetime.now(self.timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
+          self.__rt_save()
+          break
+    except Exception as e:
+      self.P(f"Error tracking response in cstore: {e}", color='r')
+    return
+
+
+  def __rt_save(self):
+    """Write recent requests to chainstore."""
+    self.chainstore_hset(
+      hkey=self.__rt_cstore_hkey,
+      key=self.ee_id,
+      value=list(self.__rt_recent_requests),
+    )
+    return
+
+
+  def _maybe_log_tracked_requests(self):
+    """Call from process(). Periodically logs cross-node request data."""
+    if self.__rt_recent_requests is None:
+      return
+    if (self.time() - self.__rt_last_log_time) > self.__rt_log_interval:
+      try:
+        hkey = self.__rt_cstore_hkey
+        all_requests = self.chainstore_hgetall(hkey=hkey)
+        if all_requests:
+          self.P(f"{hkey} requests across all nodes:\n{self.json_dumps(all_requests, indent=2)}")
+        else:
+          self.P(f"{hkey} requests across all nodes: no data")
+      except Exception as e:
+        self.P(f"Error dumping requests from cstore: {e}", color='r')
+      self.__rt_last_log_time = self.time()
+    return

ver.py

@@ -1 +1 @@
-__VER__ = '2.10.67'
+__VER__ = '2.10.68'