Merge pull request #361 from Ratio1/develop

feat: save deeploy requests in cstore

Author: cristibleotiu

extensions/business/deeploy/deeploy_manager_api.py

@@ -20,6 +20,8 @@
 
 from naeural_core.business.default.web_app.supervisor_fast_api_web_app import SupervisorFastApiWebApp as BasePlugin
 
+DEEPLOY_REQUESTS_CSTORE_HKEY = "DEEPLOY_REQUESTS"
+DEEPLOY_REQUESTS_MAX_RECORDS = 5
 
 __VER__ = '0.6.0'
 
@@ -38,6 +40,7 @@
   'WARMUP_DELAY' : 300,
   'PIPELINES_CHECK_DELAY' : 300,
   'MIN_ETH_BALANCE' : 0.00005,
+  'REQUESTS_LOG_INTERVAL' : 5 * 60,
 
   'VALIDATION_RULES': {
     **BasePlugin.CONFIG['VALIDATION_RULES'],
@@ -82,8 +85,41 @@ def on_init(self):
         color='r', boxed=True
       )
       self.maybe_stop_tunnel_engine()
+    # Request tracking state
+    self.__recent_requests = self.deque(maxlen=DEEPLOY_REQUESTS_MAX_RECORDS)
+    self.__last_requests_log_time = 0
     return
-  
+
+
+  def on_request(self, request):
+    """
+    Hook called when a new request arrives from the FastAPI side.
+    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')
+      endpoint = value[0] if isinstance(value, (list, tuple)) and len(value) > 0 else 'unknown'
+      record = {
+        'ts': self.datetime.now(self.timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
+        'endpoint': endpoint,
+      }
+      self.__recent_requests.append(record)
+      self.chainstore_hset(
+        hkey=DEEPLOY_REQUESTS_CSTORE_HKEY,
+        key=self.ee_id,
+        value=list(self.__recent_requests),
+      )
+    except Exception as e:
+      self.P(f"Error tracking request in cstore: {e}", color='r')
+    return
+
+
   def __check_eth_balance(self):
     """
     Check if the oracle has enough ETH to cover gas fees for web3 transactions.
@@ -1045,4 +1081,17 @@ 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
     return

ver.py

@@ -1 +1 @@
-__VER__ = '2.10.50'
+__VER__ = '2.10.60'